Mario Bravetti

Marco Carbone

Thomas Hildebrandt

Ivan Lanese

Jacopo Mauro

Jorge A Perez

Gianluigi Zavattaro

Jorge A Pérez

Towards Global and Local Types for Adaptation

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

Introduction

Modern complex distributed software systems face the great challenge of adapting to varying contextual conditions, user requirements or execution environments. Serviceoriented Computing (SOC), and service-oriented architectures in general, have been designed to support a specific form of adaptation: services can be dynamically discovered and properly combined in order to achieve an overall service composition that satisfies some specific desiderata that could be known only at service composition time. Rather sophisticated theories have been defined for checking and guaranteeing the correctness of these service assemblies (see, e.g., the rich literature on choreography/orchestration languages [START_REF] Bravetti | Towards a unifying theory for choreography conformance and contract compliance[END_REF][START_REF] Lanese | Bridging the gap between interaction-and process-oriented choreographies[END_REF], behavioral contracts [START_REF] Castagna | A theory of contracts for web services[END_REF][START_REF] Carpineti | A basic contract language for web services[END_REF], and session types [START_REF] Carbone | Structured communication-centered programming for web services[END_REF][START_REF] Honda | Multiparty asynchronous session types[END_REF][START_REF] Carbone | Deadlock-freedom-by-design: multiparty asynchronous global programming[END_REF]). In this paper, we consider a more fine-grained form of adaptation that can occur when the services have been already combined but have not yet completed their task. This form of adaptation may arise, for instance, when the desiderata dynamically change or when some unexpected external event occurs. In particular in the context of computer-supported case management, e.g. for health-care or financial services, changes are the norm rather than the exception. This has lead to an increasing interest both from academia and industry in the development of technologies supporting dynamic changes in choreographies and processes, collectively referred to as adaptive case management (ACM) [START_REF] Swenson | Mastering the Unpredictable -How Adaptive Case Management Will Revolutionize the Way That Knowledge Workers Get Things Done[END_REF][START_REF] Mukkamala | Towards trustworthy adaptive case management with dynamic condition response graphs[END_REF] and being addressed in the recent proposal for a Case Management Model and Notation (CMMN) from OMG [START_REF] Omg | Case management model and notation 1.0 -beta 1[END_REF]. For such technologies, it is crucial that modifications occur in a consistent and coordinated manner in order to avoid breaking the correctness of the overall service composition.

In this paper, we initiate the investigation of new models and theories for service composition that properly take into account this form of adaptation. First of all, we extend a previous language for the description of service choreographies [START_REF] Bravetti | Towards a unifying theory for choreography conformance and contract compliance[END_REF] with two operators: the first one allows for the specification of adaptable scopes that can be dynamically modified, while the second may dynamically update code in one of such scopes. This language is designed for the global description of dynamically adaptable multi-party interaction protocols. As a second step in the development of our theory, we define a service behavioral contract language for the local description of the inputoutput communications. In order to support adaptation, also in this case we enhance an existing service contract language [START_REF] Bravetti | Towards a unifying theory for choreography conformance and contract compliance[END_REF] with two new operators for adaptable scope declaration and code update, respectively. The most challenging aspect to be taken into account is the fact that, at the local level, peers should synchronize their local adaptations in order to guarantee a consistent adaptation of their behavior. As mentioned above, these two languages are expected to be used to describe multi-party protocols from global and local perspectives, respectively. The relationship between the two languages is formalized in terms of a projection function which allows us to obtain endpoint specifications from a given choreography.

The complete theory that we plan to develop will also consider a concrete language for programming services; such a language will include update mechanisms like those provided by, for instance, the Jorba service orchestration language [START_REF] Lanese | A framework for rule-based dynamic adaptation[END_REF]. The ultimate aim of our research is to define appropriate behavioral typing techniques able to check whether the concretely programmed services correctly implement the specified multiparty adaptable protocols. This will be achieved by considering the global specification of the protocol, by projecting such specification on the considered peer, and then by checking whether the actual service correctly implements the projected behavior. In order to clarify our objective, we discuss an example inspired by a health-care scenario [START_REF] Lyng | From paper based clinical practice guidelines to declarative workflow management[END_REF]. Two adaptable protocols are described by using the proposed choreography languages: the first protocol describes the interaction between the doctor and the laboratory agents, while the second involves a doctor, a nurse, and a patient. In case of emergency, the doctor may speed up the used protocols by interrupting running tests and avoiding the possibility that the nurse refuses to use a medicine she does not trust -this possibility is normally allowed by the protocol. Then, using a π-calculus-like language, we present the actual behavior of the doctor and discuss the kinds of problems that we will have to address in order to define appropriate behavioral type checking techniques.

Structure of the paper. The next section introduces choreography and endpoint languages with adaptation constructs, and the projection function that relates global and local specifications. Then, in §3 we outline a concrete specification language and discuss the health-care scenario. In §4 we present some concluding remarks and briefly review related works.

Disclaimer. This paper discusses ongoing work supported by the "Behavioural Types for Reliable Large-Scale Software Systems" (BETTY) Cost Action. Our main aim is to report about the current state of this research activity.

In the paper, we use the following sets: channels, ranged over by a, a ′ , . . .; scope names, ranged over by X, X ′ , . . .; and roles/participants, ranged over by r, r 1 , r 2 , . . .. Also, we use T, T ′ , . . . to denote sets of roles.

Choreography Language

Syntax. We describe here the syntax of our choreography language. To this end, we first define a set of so-called choreography terms. Then, by requiring some well-formedness conditions on such terms, we obtain actual choreographies.

The syntax of choreography terms is as follows:

C ::= a r1→r2 (interaction) | C ; C (sequence) | C | C (parallel) | C + C (choice) | C * (star) | 1 (one) | 0 (nil) | X : T [C] (scope) | X r {C} (update)
The basic element of a choreography term C is an interaction a r1→r2 , with the intended meaning that participant r 1 sends a message to participant r 2 over channel a. Two terms

C 1 and C 2 can be composed in sequence (C 1 ; C 2 ), in parallel (C 1 | C 2 )
, and using nondeterministic choice (C 1 + C 2 ). Also, a choreography term may be iterated zero or more times using the Kleene star * . The empty choreography term, which just successfully terminates, is denoted by 1. The deadlocked choreography term 0 is needed for the definition of the semantics: we will assume that it is never used when writing a choreography (see Definition 1).

The two last operators deal with adaptation. Adaptation is specified by defining a scope that delimits a choreography term that, at runtime, may be replaced by a new choreography term, coming from either inside or outside the system. Adaptations coming from outside may be decided by the user through some adaptation interface, by some manager module, or by the environment. In contrast, adaptations coming from inside represent self-updates, decided by a part of the system towards itself or towards another part of the system, usually as a result of some interaction producing unexpected values. Adaptations from outside and from inside are indeed quite similar, e.g., an update decided by a manager module may be from inside if the manager behavior is part of the choreography term, from outside if it is not. Construct X : T [C] defines a scope named X currently executing choreography term C -the name is needed to designate it as a target for a particular adaptation. Type T is the set of roles (possibly) occurring in the scope. This is needed since a given update can be applied to a scope only if it specifies how all the involved roles are adapted. Operator X r {C} defines internal updates, i.e., updates offered by a participant of the choreography term. Here r denotes the participant offering the update, X is the name of the target scope, and C is the new choreography term.

Not all choreography terms generated by the syntax above are useful choreographies. To formally define the choreography terms which actually represent choreographies, we rely on some auxiliary definitions. The set of roles inside a choreography term C, denoted roles(C), is defined inductively as follows:

roles(a r1→r2 ) = {r 1 , r 2 } roles(X r {C}) = {r} roles(X : T [C]) = T ∪ roles(C) roles(C * ) = roles(C) roles(C 1 ; C 2 ) = roles(C 1 | C 2 ) = roles(C 1 + C 2 ) = roles(C 1 ) ∪ roles(C 2 ) roles(1) = roles(0) = ∅
Notice that for X r {C} we consider role r but not the roles in C. This is because X r {C} may correspond to an external update on some different choreography term. We are now ready to define choreographies.

Definition 1 (Choreography). A choreography term C is a choreography if:

1. C does not contain occurrences of 0; 2. all names of scopes in C are pairwise distinct; 3. C is well-typed, i.e. for every scope X :

T [C ′ ] occurring in C: -roles(C ′ ) ⊆ T and -every update prefix X r {C ′′ } occurring in C is such that roles(C ′′ ) ⊆ T .
We use type(X) to denote the type T associated to the unique scope X :

T [C ′ ].
Semantics. We now define the semantics of choreography terms via a labeled transition system. As in the syntax, the most interesting part of the semantics concerns update constructs. Recall that T is a set of roles. In the definition below, we use C[C ′ /X] to denote the substitution that replaces all scopes X : T [C ′′ ] with name X occurring in C (not inside update prefixes) with X :

T [C ′ ].
As usual, transition C α -→ C ′ intuitively says that choreography term C may evolve to C ′ by performing an action represented by a label α. Our set of labels includes √ (termination), a r1→r2 (interaction), and X r {C} (update).

Definition 2. The semantics of choreography terms is the smallest labeled transition system closed under the rules in Table 1.

We briefly comment on the rules in Table 1. Rules in the first four rows of the table are standard (cf. [START_REF] Bravetti | Towards a unifying theory for choreography conformance and contract compliance[END_REF]). Rule (ONE) defines termination for the empty choreography term. Rule (COMM) executes an interaction, making it visible in the label. While rule (SEQ) allows the first component of a sequential composition to compute, rule (SEQTICK) allows it to terminate, starting the execution of the second component. Rule (PAR) allows parallel components to interleave their executions. Rule (PARTICK) allows parallel components to synchronize their termination. Rule (CHO) selects a branch in a nondeterministic choice. Rule (STAR) unfolds the Kleene star. Note that the unfolding may break uniqueness of scopes with a given name-we will come back to this point later on. Rule (STARTICK) defines termination of a Kleene star.

The remaining rules in Table 1 deal with adaptation. Rule (COMMUPD) makes an internal adaptation available, moving the information to the label. Adaptations propagate through sequence, parallel composition, and Kleene star using rules (SEQUPD), 1. Semantics of Choreography Terms (PARUPD), and (STARUPD), respectively. Note that, while propagating, the update is applied to the continuation of the sequential composition, to parallel terms, and to the body of Kleene star. Notably, the update is applied to both enabled and non enabled occurrences of the desired scope. Rule (SCOPEUPD) allows a scope to update itself (provided that the names coincide), while propagating the update to the rest of the choreography term. Rule (SCOPE) allows a scope to compute.

(ONE) 1 √ -→ 0 (COMM) ar 1 →r 2 ar 1 →r 2 -----→ 1 (SEQ) C1 ar 1 →r 2 -----→ C ′ 1 C1; C2 ar 1 →r 2 -----→ C ′ 1 ; C2 (SEQTICK) C1 √ -→ C ′ 1 C2 α -→ C ′ 2 C1; C2 α -→ C ′ 2 (PAR) C1 ar 1 →r 2 -----→ C ′ 1 C1 | C2 ar 1 →r 2 -----→ C ′ 1 | C2 (PARTICK) C1 √ -→ C ′ 1 C2 √ -→ C ′ 2 C1 | C2 √ -→ C ′ 1 | C ′ 2 (CHO) C1 α -→ C ′ 1 C1 + C2 α -→ C ′ 1 (STAR) C ar 1 →r 2 -----→ C ′ C * ar 1 →r 2 -----→ C ′ ; C * (STARTICK) C * √ -→ 0 (COMMUPD) Xr{C} Xr {C} ----→ 1 (SEQUPD) C1 Xr {C} ----→ C ′ 1 C1; C2 Xr {C} ----→ C ′ 1 ; (C2[C/X]) (PARUPD) C1 Xr {C} ----→ C ′ 1 C1 | C2 Xr {C} ----→ C ′ 1 | (C2[C/X]) (STARUPD) C1 Xr {C} ----→ C ′ 1 C * 1 Xr {C} ----→ C ′ 1 ; (C1[C/X]) * (SCOPEUPD) C1 Xr {C} ----→ C ′ 1 X : T [C1] Xr {C} ----→ X : T [C] (SCOPE) C1 α -→ C ′ 1 α = Xr{C} for any r, C X : T [C1] α -→ X : T [C ′ 1 ] Table
We can now define the notion of closed traces that correspond to computations of stand-alone choreography terms.

Definition 3 (Traces). Given a choreography term

C 0 a trace is a (possibly infinite) sequence C 0 α1 -→ C 1 α2 -→ C 2 α3 -→ • • • .
In order to model choreography terms that can be externally updated we need to introduce the notion of open transitions. ---→ C i+1 is either a transition of the semantics in Table 1 or an open transition.

Definition 4 (Open transitions). The choreography term C has an open transition of the form

C X{C ′′ } -----→ C[C ′′ /X] if: -there is a choreography C 0 with a trace C 0 α -→ • • • α ′ -→ C ′ 0 |C; -C ′ 0 Xr{C ′′ } -----→ C
As we have said, in a choreography we assume scope names to be unique. However, uniqueness is not preserved by transitions. Nevertheless a slightly weaker property (arising from the fact that we consider Kleene star as the only form of recursion) is indeed preserved, and it simplifies the implementation of the adaptation mechanisms at the level of endpoints.

Proposition 1. Let C be a choreography and let C ′ be a choreography term reachable from C via zero or more transitions (possibly open). For every X, C ′ contains at most one occurrence of a scope named X which is enabled (i.e., which can compute).

An Example. Below we give an example of an adaptable choreography to illustrate the features introduced above. The example is based on a health-care workflow inspired by field study [START_REF] Lyng | From paper based clinical practice guidelines to declarative workflow management[END_REF] carried out in previous work. The field study was also considered as inspiration for recent work on session types for health-care processes [START_REF] Henriksen | Trustworthy pervasive healthcare services via multiparty session types[END_REF] and adaptable declarative case management processes [START_REF] Mukkamala | Towards trustworthy adaptive case management with dynamic condition response graphs[END_REF], but the combination of session types and adaptability has not been treated previously.

In the considered scenario, doctors, nurses and patients employ a distributed, electronic health-care record system where each actor (including the patient) uses a tablet pc/smartphone to coordinate the treatment. Below, iteration C + stands for C; C * .

X : {D, N }[ (prescribe D→N ) + ; (sign D→N + X D {sign D→N }; up D→N ); trust N →D + ]; medicine N →P
where D, N, P denote participant doctors, nurses, and patients, respectively.

The doctor first records one or more prescriptions, which are sent to the nurse's tablet (prescribe D→N ) + . When receiving a signature, sign D→N , the nurse informs the doctor if the prescription is trusted. If not trusted then the doctor must prescribe a new medicine. If trusted, the nurse proceeds and gives the medicine to the patient, which is recorded at the patient's smartphone, medicine N →P . However, instead of signing and waiting for the nurse to trust the medicine, in emergency cases the doctor may update the protocol so that the possibility of not trusting the prescription is removed: the nurse would have to give the medicine to the patient right after receiving the signature. In the example, this is done by a self-update (X D {sign D→N }) of the running scope. In other scenarios, this could have been done by an entity not represented in the choreography, such as the hospital director, thus resulting in an external update. The doctor notifies the protocol update to the nurse using the up D→N interaction. Now consider the further complication that the doctor may run a test protocol with a laboratory, after prescribing a medicine and before signing:

X ′ {D, L} : [orderT est D→L ; (results L→D + X ′ D {1})]
We allow the test protocol also to be adaptable, since the doctor may decide that there is an emergency while waiting for the results, and thus also having to interrupt the test protocol. If the two protocols are performed in interleaving by the same code, then the updates of the two protocols should be coordinated. We illustrate this in § 3 below.

Endpoint Language

Since choreographies are at the very high level of abstraction, defining a description of the same system nearer to an actual implementation is of interest. In particular, for each participant in a choreography (also called endpoint) we would like to describe the actions it has to take in order to follow the choreography. The syntax of endpoint processes is as follows:

P ::= a r (output) | a r (input) | P ; P (sequence) | P | P (parallel) | P + P (choice) | P * (star) | 1 (one) | 0 (zero) | X[P ] F (scope) | X (r1,...,rn) {P 1 , . . . , P n } (update)
where F is either A, denoting an active (running) scope, or ε, denoting a scope still to be started (ε is omitted in the following).

As for choreographies, endpoint processes contain some standard operators and some operators dealing with adaptation. Communication is performed by a r , denoting an output on channel a towards participant r. Dually, a r denotes an input from participant r on channel a. Intuitively, an output a r in role s and an input a s in role r should synchronize. Two endpoint processes P 1 and P 2 can be composed in sequence (P 1 ; P 2 ), in parallel (P 1 | P 2 ), and using nondeterministic choice (P 1 + P 2 ). Endpoint processes can be iterated using a Kleene star * . The empty endpoint process is denoted by 1 and the deadlocked endpoint process is denoted by 0.

Adaptation is applied to scopes. X[P ] F denotes a scope named X executing process P . F is a flag distinguishing scopes whose execution has already begun (A) from scopes which have not started yet (ε). The update operator X (r1,...,rn) {P 1 , . . . , P n } provides an update for scope named X, involving roles r 1 , . . . , r n . The new process for role r i is P i .

Endpoints are of the form [[P ]] r , where r is the name of the endpoint and P its process. Systems, denoted S, are obtained by composition of parallel endpoints:

S ::= [[P ]] r (endpoint) | S||S (parallel system)
As for choreographies, not all systems are endpoint specifications. By a slight abuse of notation we extend type(X) to endpoints associating a set of roles to each scope name X. Endpoint specifications are defined as follows. Definition 6. A system S is an endpoint specification if the following conditions hold: (i) no active scopes are present (ii) endpoint names are unique (iii) all roles r occurring in terms of the form a r , a r , or such that r ∈ type(X) for some scope X are endpoints of S (iv) a scope with name X con occur (outside updates) only in endpoints r ∈ type(X) (v) every update has the form X type(X) {P 1 , . . . , P n } (vi) outputs a r and inputs a r included in X type(X) {P 1 , . . . , P n } are such that r ∈ type(X).

In this presentation, we do not formally define a semantics for endpoints: we just point out that it should include labels corresponding to all the labels of the semantics of choreography terms, plus some additional labels corresponding to partial activities, such as an input. We also highlight the fact that all scopes which correspond to the same choreography scope evolve together: their scope start transitions (transforming a scope from inactive to active) are synchronized, as well as their scope end transitions (removing it). The fact that choreographies feature at most one scope with a given name is instrumental in ensuring this property.

Projection

Since choreographies provide system descriptions at the high level of abstraction and endpoint specifications provide more low level descriptions, a main issue is to derive from a given choreography an endpoint specification executing it. This is done using the notion of projection. Definition 7 (Projection). The projection of a choreography C on a role r, denoted by C ↾ r , is defined by the clauses below

a r1→r2 ↾ r =      a r2 if r = r 1 a r1 if r = r 2 1 otherwise X r ′ {C} ↾ r = X (r1,...,rn) {C ↾ r1 , . . . , C ↾ rn }with {r 1 , . . . , r n } = type(X) if r = r ′ 1 otherwise X : T [C] ↾ r = X[C ↾ r ] if r ∈ type(X)
1 otherwise
and is an homomorphism on the other operators. The endpoint specification resulting from a choreography C is obtained by composing in parallel roles [[C ↾ r ]] r , where r ∈ roles(C).

As an example, the endpoint projection obtained from the prescribe choreography introduced in §2.1 is

[[P N ]] N ||[[P D ]] D ||[[P P
]] P where processes P N , P D , and P P are as fol-lows (we omit unnecessary 1 processes):

P N = X[((prescribe D ) + ; (sign D + up D ); trust D ) + ] ; medicine P P D = X[((prescribe N ) + ; (sign N + X D,N {sign N , sign D } ; up N ); trust N ) + ] P P = medicine N
One can see that the system S obtained by projecting a choreography is an endpoint specification. Ideally, traces of the projected system should correspond to the traces of the original choreography. Actually, we conjecture that this occurs only for choreographies satisfying suitable connectedness conditions that we plan to formalize extending those in [START_REF] Lanese | Bridging the gap between interaction-and process-oriented choreographies[END_REF]. This is not an actual restriction, since choreographies that do not respect the conditions can be transformed into choreographies that respect them [START_REF] Lanese | Amending choreographies[END_REF].

Conjecture 1. Traces of projection of connected choreographies correspond to traces of the original choreography.

We point out two main aspects of the correspondence. First, labels X r {C} of transitions of the choreography should be mapped to labels [X (r1,...,rn) {P 1 , . . . , P n }] r of the transitions of the endpoint specification, where type(X) = {r 1 , . . . , r n } and P 1 = C ↾ r1 , . . . , P n = C ↾ rn are obtained by projection from C. Second, endpoint traces should not consider unmatched input and output labels.

Typing a Concrete Language

As demonstrated by our examples, choreography and endpoint terms provide a useful language for expressing protocols with adaptation. In this section, we investigate the idea of using such protocols as specifications for a programming language with adaptation. We plan to follow the approach taken in multiparty session types [START_REF] Honda | Multiparty asynchronous session types[END_REF], where choreographies (and endpoints) are interpreted as behavioral types for typing sessions in a language modeled as a variant of the π-calculus. In the sequel, we investigate the core points of such a language by giving an implementation that uses the protocols specified in the examples of the previous sections. In particular, we discuss what are the relevant aspects for developing a type system for such a language, whose types are the choreographies introduced in § 2.1.

In both prescribe and test protocols, the doctor plays a key role since (s)he initiates the workflow with prescriptions, decides when tests have to be requested, and decides when the protocols have to be interrupted due to an emergency. A possible implementation of the doctor could be given by the following program:

1. P D = pr(k); X[ repeat {repeat { 2.
k : prescribe N e pr ; test(k ′ );

3.

X ′ [k ′ : orderT est L e o ; 4.

(k ′ : results L (x)+ 5.

X ′ (D,L) {X (D,N ) {k : sign N e s , k : sign D (z)}, 1})] 6.

} until ok(x); k : trust N (t)} until trusted(t) ]

In the code there are two kinds of communication operations, namely protocol initiation operations, where a new protocol (or session) is initiated, and in-session operations where protocol internal operations are implemented. The communication pr(k) is for initiating a protocol called pr and its semantics is to create a fresh protocol identifier k that corresponds to a particular instance of protocol pr. In-session communications are standard.

The novelty in the process above is in the scope X[. . .] and update X ... {. . .}, which state respectively that the program can be adapted at any time in that particular point, and that an adaptation is available. Interestingly enough, the way the program P D uses the protocols needs care. If the doctor wants to adapt to emergency while waiting for tests, both the test protocol and the prescription protocol need to be adapted as shown in line 5. If the doctor adapts to emergency after having received tests that are ok, then only the prescription protocol needs to be adapted. One can see that session pr can be typed using the prescribe endpoint specification and session test using the test endpoint specification. The update of X in line 4 does not appear in the protocol test since it acts as an external update for a different protocol.

Concluding Remarks and Related Work

Adaptation is a pressing issue in the design of service-oriented systems, which are typically open and execute in highly dynamic environments. There is a rather delicate tension between adaptation and the correctness requirements defined at service composition time: we would like to adapt the system's behavior whenever necessary/possible, but we would also like adaptation actions to preserve overall correctness.

In this paper, we have reported ongoing work on adaptation mechanisms for serviceoriented systems specified in terms of choreographies. By enhancing an existing language for choreographies with constructs defining adaptation scopes and dynamic code update, we obtained a simple, global model for distributed, adaptable systems. We also defined an endpoint language for local descriptions, and a projection mechanism for obtaining (low-level) endpoint specifications from (high-level) choreographies.

We now briefly comment on related works. The work in [START_REF] Dalla Preda | Safe run-time adaptation of distributed applications[END_REF] is closely related, and indeed was a source of inspiration for the current work. It develops a framework for rule-based adaptation in a choreographic setting. Both choreographies and endpoints are defined; their relation is formally defined via projection. The main difference w.r.t. the work described here is our choice of expressing adaptation in terms of scopes and code update constructs, rather than using rules. Also, we consider choreographies as types and we allow multiple protocols to interleave inside code. These problems are not considered in [START_REF] Dalla Preda | Safe run-time adaptation of distributed applications[END_REF].

Our work is also related to the recent work [START_REF] Coppo | Self-adaptive monitors for multiparty sessions[END_REF], which considers self-adaptive systems monitored by different global descriptions. The description specifies also when the used monitor should change, and the new monitor to be used is determined by an adaptation function. A main difference is that in their approach the code does not change because processes should be able to implement all the global descriptions since the very beginning.

Our approach bears some similarities with works on multiparty sessions [START_REF] Honda | Multiparty asynchronous session types[END_REF][START_REF] Carbone | Structured communication-centered programming for web services[END_REF], and in particular with works dealing with exceptions in multiparty sessions [START_REF] Capecchi | Global escape in multiparty sessions[END_REF]. Our focus so far has been on formally relating global and local descriptions of choreographies via projection and trace correspondence; investigating correctness properties (e.g., communication safety) via typing in our setting is part of ongoing work. We also note that exceptions and adaptation are similar but conceptually different phenomena: while the former are typically related to foreseen unexpected behaviors in (low-level) programs, adaptation appears as a more general issue, for it should account for (unforeseen) interactions between the system and its (varying) environment.

We have borrowed inspiration also from [START_REF] Mukkamala | Towards trustworthy adaptive case management with dynamic condition response graphs[END_REF], in which adaptive case management is investigated via Dynamic Condition Response (DCR) Graphs, a declarative process model.

Finally, the adaptation constructs we have considered for choreographies and endpoints draw inspiration from the adaptable processes defined in [START_REF] Bravetti | Adaptable processes[END_REF]. The application of adaptable processes in session-typed models of structured communications (focusing on the case of binary sessions) has been studied in [START_REF] Di Giusto | Disciplined structured communications with consistent runtime adaptation[END_REF].

An immediate topic for future work is the full formalization of the concrete language and its typing disciplines. Other avenues for future research include the investigation of refinement theories with a testing-like approach, enabled by having both systems and adaptation strategies modeled in the same language, and the development of prototype implementations.

′′0-→ C 1 α2-→ C 2 α3-

 12 where r ∈ roles(C) and X is the name of a scope in C. We can now define the notion of open traces corresponding to computations including also open transitions. Definition 5 (Open Traces). Given a choreography term C 0 an open trace is a (possibly infinite) sequence C 0 α1 → • • • where every C i αi+1

  sign N e s + X (D,N ) {k : sign N e s , k : sign D (z)} ; k : up N ); 8.

Acknowledgments. This work was partially supported by COST Action IC1201: Behavioural Types for Reliable Large-Scale Software Systems (BETTY). Jorge A. Pérez was partially supported by grants SFRH / BPD / 84067 / 2012 and CITI of the Portuguese Foundation for Science and Technology (FCT).


Radu Tudoran
email: radu.tudoran@inria.fr

Kate Keahey
email: keahey@mcs.anl.gov

Pierre Riteau
email: priteau@uchicago.edu

Sergey Panitkin
email: panitkin@bnl.gov

Gabriel Antoniu
email: gabriel.antoniu@inria.fr

Evaluating Streaming Strategies for Event Processing across Infrastructure Clouds

Infrastructure clouds revolutionized the way in which we approach resource procurement by providing an easy way to lease compute and storage resources on short notice, for a short amount of time, and on a pay-as-you-go basis. This new opportunity, however, introduces new performance trade-offs. Making the right choices in leveraging different types of storage available in the cloud is particularly important for applications that depend on managing large amounts of data within and across clouds. An increasing number of such applications conform to a pattern in which data processing relies on streaming the data to a compute platform where a set of similar operations is repeatedly applied to independent chunks of data. This pattern is evident in virtual observatories such as the Ocean Observatory Initiative, in cases when new data is evaluated against existing features in geospatial computations or when experimental data is processed as a series of time events. In this paper, we propose two strategies for efficiently implementing such streaming in the cloud and evaluate them in the context of an ATLAS application processing experimental data. Our results show that choosing the right cloud configuration can improve overall application performance by as much as three times.

I. INTRODUCTION

By providing a way to lease compute and storage resources on demand, for a short amount of time, and on a pay-as-you-go basis, infrastructure cloud computing revolutionized the way in which we approach resource procurement. This innovation is proving fundamental to the construction of systems capable of rapid scaling and high reliability. At the same time, however, as has been shown by multiple evaluation efforts [START_REF] Yelick | The Magellan Report on Cloud Computing for Science[END_REF], [START_REF] Juve | An evaluation of the cost and performance of scientific workflows on amazon ec2[END_REF], [START_REF] Sakr | A survey of large scale data management approaches in cloud environments[END_REF], the performance trade-offs inherent in current virtualization technology mean that data-intensive applications are often not the best fit for infrastructure clouds and consequently cannot leverage the advantages of cloud computing.

This technological shortcoming is particularly impactful in the Big Data era. An increasing number of Big Data applications conform to a pattern in which data processing relies on streaming the data to a compute platform where a set of similar operations is repeatedly applied to independent chunks of data. This pattern is evident in virtual observatories such as the Ocean Observatory Initiative [START_REF]Ocean Observatory Initiative[END_REF], in cases when new data is evaluated against existing features in geospatial computations (e.g., FluMapper [START_REF] Padmanabhan | Flumapper: an interactive cybergis environment for massive location-based social media data analysis[END_REF]) or when experimental data is processed as a series of time events [START_REF] Balewski | Offloading peak processing to virtual farm by STAR experiment at RHIC[END_REF], [START_REF] Jackson | Seeking supernovae in the clouds: a performance study[END_REF]. These applications often need real-time response time, for example, to adaptively redeploy sensors in a virtual observatory or to support an experiment. Moreover, they need to scale rapidly to fluctuating request numbers as the detection of new phenomena generates more questions and thus necessitates more processing. This type of application clearly could benefit from on-demand resource availability provided by cloud computing, but they are often costly or hard to structure because of difficulties and inefficiencies in data management in the cloud.

In this paper, we propose two strategies for efficiently implementing streaming in the cloud for such applications. The first strategy seeks to overlap computation and communication by streaming data directly to the nodes where the computation takes place, in such a way that the rate of data streaming keeps pace with computation; the second strategy relies on first copying data to the cloud and then using it for computation. We evaluate these strategies in the context of an Atlas application application [START_REF]ATLAS[END_REF], [START_REF]ATLAS Applications[END_REF] processing experimental data in terms of both performance and cost. Our results indicate the ways in which communication in the cloud can interfere in computation. We show that choosing the right cloud configuration can improve overall application performance by as much as three times and can significantly reduce the cost.

The paper is organized as follows. Section II describes the features of infrastructure clouds relevant to this paper, defines the type of applications and the interaction pattern addressed in the paper, and describes the two streaming strategies evaluated in the paper. Section III presents our experimental setup and describes the experiments we per-formed to evaluate the two strategies. Section IV discusses their respective costs. In Section V we discuss related work. In Section VI we summarize our conclusions.

II. BACKGROUND A. Cloud Storage Basics

Infrastructure-as-a-Service (IaaS) clouds allow users to deploy virtual machines (VMs) instances, thus leasing computing capacity for a short amount of time usually paid for per hour. In addition to computing capability, cloud providers offer several types of storage with different availability, access, and performance characteristics, at different price. These types of storage are as follows.

• Ephemeral storage -is the local virtual disk attached to a deployed instance. Its storage capacity can reach up to many terabytes depending on the instance type, but it persists only for the lifetime of the instance and is thus subject to sudden loss when the instance fails unexpectedly. The storage is free in the sense that it comes with the cost of leasing an instance. • Persistent attached storage -provides networkattached block-level storage volumes that can be attached to a running instance and exposed as a storage device within that instance. The lifetime of this type of storage is independent of the instances that mount it, i.e., it can be reused by many instances. Charges vary based on the size of volumes used and period of usage. Examples of persistent attached storage are Amazon Elastic Block Storage (EBS) [START_REF]Amazon EBS[END_REF] and Azure drives [START_REF]Azure Drives[END_REF].

• Cloud storage -offers data storage as binary objects (blobs), usually in a two-level namespace structure (e.g., blobs and containers). This type of storage typically offers different levels of service, for example, in terms of access time or redundancy, at different prices. Users typically get charged based on the data size and on the storage time. Examples of cloud storage include Amazon Simple Storage Service (S3) [START_REF]Amazon S3[END_REF], Amazon Glacier [START_REF]Amazon Glacier[END_REF], Microsoft Azure BLOBs [START_REF]Microsoft Azure BLOBs[END_REF], [START_REF] Calder | Windows azure storage: a highly available cloud storage service with strong consistency[END_REF], and Google Cloud Storage [START_REF]Google Cloud Storage[END_REF]. While cloud storage has been used in the past to provide a file system to support a running application [START_REF] Jackson | Seeking supernovae in the clouds: a performance study[END_REF], this type of storage is increasingly evolving to provide an archival capability. Therefore, in this paper we focus on the first two types of storage.

B. Streaming Applications

Increasingly more applications conform to a pattern in which an operation is applied to many independent chunks of data. This can happen because an operation is applied to data regularly produced by sensors in virtual observatories such as the Ocean Observatory initiative [START_REF]Ocean Observatory Initiative[END_REF], when new data is evaluated against existing features in geospatial computations [START_REF] Kazemitabar | Geostreaming in cloud[END_REF], [START_REF] Costan | Tomus-Blobs: Scalable Data-intensive Processing on Azure Clouds[END_REF] or when experimental data is processed as a series of time events. The data typically is streamed from the data source to a cluster or a cloud, where the computation takes place, either chunk by chunk or by bunching multiple chunks together, with the streaming constituting a significant component of the overall computation. Optimizing this scenario requires streaming to keep pace with computing. In this paper, we call such applications "streaming applications", and we investigate how they can best be mapped onto infrastructure cloud storage facilities.

As an example of a streaming application we use a data analysis code from an ATLAS experiment at the Large Hadron Collider [START_REF]ATLAS[END_REF], [START_REF]ATLAS Applications[END_REF]. The application performs data analysis searches in a channel where the Higgs decays into t anti-t quarks. The experimental data is collected as successive time events, one event corresponding to the aggregated readings from the ATLAS sensors at a given moment. Because the amount of collected data is on the order of tens of petabytes, efficient processing is of significant concern.

The application is written in the ROOT [START_REF]ROOT Framework[END_REF] version of C and is executed as a workflow using the ProofLite framework. The typical computation pattern consists of reading the job and data description, followed by successive reads of events that need to be processed. The data has been divided into independent chunks by annotating it with metadata, thus eliminating the need for communication between processes but adding to the bulk of data that must be streamed to each worker. The data-streaming technology used to run the application is CERN's xrootd client-server framework.

C. Streaming Scenarios

We consider two scenarios for managing data streaming on infrastructure cloud facilities. In each scenario we work with a data source and then deploy a set of compute instances (or compute VMs) on a remote cloud. Each compute instance is running a number of application workers that process the data. The architecture of these scenarios is presented in Figure 1.

• Stream&Compute -In this scenario data is streamed directly from the data source to the compute instances where it is ingested and processed by the worker processes. The events are processed in memory as they are streamed.

• Copy&Compute -In this scenario we allocate some persistent storage and an additional instance that mounts this storage in order to make it accessible to all compute instances. The streaming is split into two phases. First, we copy the data from the data source to this persistent attached storage. Second, once all data is available locally on the cloud, it is streamed locally to the compute instances so that the computation can begin. The advantages of Stream&Compute are that (1) it provides better response time on a case by case basis where not all of the data chunks to be computed are available up front or need to be processed to yield a result; (2) it has the potential to overlap computation and communication, thus potentially shortening the time to compute a group of events; and (3) it uses storage provided with instances, thus making the computation potentially cheaper. At the same time, we note that at large scales, network saturation can slow the data-streaming rate to the point where it no longer keeps pace with computation potentially necessitating redistribution across different clouds.

The advantage of Copy&Compute is that it relies on persistent storage, thus leading to easier repair in cases where an instance is terminated unexpectedly: while any terminated computations will have to be rerun, the data will not have to be resent over wide-area network.

III. EXPERIMENTS

The goal of the experiments is to assert the validity of our hypothesis and to determine the best options for processing stream data using the cloud. To this purpose, we analyze a series of experiments that compare the two main scenarios for using the cloud: Copy&Compute and Stream&Compute. We also discuss questions regarding scalability, bottleneck limits, the relation between the CPU and throughput, and the choice of instance types.

A. Experimental setup

The experiments presented here were run on FutureGrid [START_REF]FutureGrid[END_REF], using Hotel (a University of Chicago cloud configured with Nimbus version 2.10.1 [START_REF]Nimbus[END_REF]) and Sierra (a San Diego Supercomputing Center cloud configured with OpenStack Grizzly [START_REF]OpenStack[END_REF]), as well as on the Azure commercial cloud [START_REF]Azure[END_REF]. To implement persistent attached storage we leveraged OpenStack Block Storage volumes for OpenStack and Azure drives [START_REF]Azure Drives[END_REF] for Azure. All VMs used in the experiments run a CentOS 6.3 operating system. The Small instances used in FutureGrid offer 1 virtual CPU, 2 GB of memory, and 20 GB of local storage. In Azure the Small VM roles have 1 CPU, which is guaranteed to be mapped to an unshared physical CPU, 1.75 GB memory, 200 GB local storage, and 100 Mbps of bandwidth (note that while this is the advertised bandwidth, our experience is that the effective bandwidth is higher than 100 Mbps). Medium instances provide twice as many resources as do Small instances (2 virtual CPU, 4 GB memory, and 40 GB ephemeral disk); and, similarly, Large instances are 4 times larger than Small instances. The interconnection network between the Sierra and Hotel sites is a 10 Gbps network; the physical nodes are connected via a 1 Gbps Ethernet network.

The Hotel Nimbus cloud was used to host the data sources (i.e., the nodes that hold the data) for all experiments. The Sierra OpenStack cloud was used to host the compute nodes for all FutureGrid experiments. The Azure West US Data Center (located in California) was used to host the compute nodes for all Azure experiments.

The application we experimented with is composed of small data computation units of similar size, referred to as events. Small size differences appear because the events aggregate measurements of different aspects of a phenomenon, which may or may not be detected at a given point. The number of events processed is therefore a good measure of the progress of the application. We used the following metrics for our experiments.

• Compute Rate -this is the amount of events that are processed in a time unit; the metric unit is events/second. • Data Rate -this is the amount of data that is read or acquired in a time unit; the metric unit is megabytes/second. For the Stream&Compute case, we measured the compute rate by running the program with a set number of events and measuring on each compute node how long it took from start to finish of the computation. For the Copy&Compute case, the experiment involved two phases: a remote site copy followed by the computation on local data. For the first phase, we used the "time" command to measure how long a remote copy took using "scp". This was then added to the time taken by the application to complete processing over the set of events transferred. We measured the data rate by dividing the amount of input data by the total time to complete its processing. For the Copy&Compute case, this total time included both the remote data copy and running the application. We increased the input data size as we scaled up the number of instances (i.e., the amount of data processed with 32 VMs is equal to 32 times the amount of data for one VM).

For each measurement presented in the charts, 100 independent experiments were performed. The values shown represent the averages.

B. Copy&Compute vs. Stream&Compute

We first present average per VM compute rates and per VM data rates to application VMs deployed on FutureGrid small instances as well as Azure small instances. The average data rate per VM is computed by taking the total data rate for a given application VM configuration and dividing it by the number of VMs on which the application was deployed. Figure 2 shows the results for FutureGrid and Figure 3 for Azure. In addition to the data shown in the figures, we note that the results for the Copy&Compute operation show a higher variability, having a coefficient of variation (i.e., standard deviation / mean) of ∼20%, than do the results for the Stream&Compute data, which has a coefficient of variation of ∼10%. The remote copy phase is mainly responsible for this high variability; the variability on local dissemination is very low. We see that in both cases Copy&Compute outperforms Stream&Compute. The Copy&Compute strategy offers three to four times better performance than does Stream&Compute. This is contrary to our expectation; we expected the Stream&Compute method to be faster because of overlapping computation and communication. The overlapping of the computation with communication is available at the level of the framework, with no particular optimizations set up on our side. Hence, similar results are expected for any framework that provides such a behavior: acquiring the data for the next step of the computation while the available data for the current step is being processed.

In addition, we see a drop in per VM average data and compute rates for the Copy&Compute case on Azure once we reach 16 VMs. This reflects the bandwidth limitation of 100 Mbps to the Azure EBS node from which data is shared among the application nodes at a potential throughput of about 4 MB/s. As more nodes want to access the data on the EBS node at the same time the maximum bandwidth of this node is reached, and the resulting contention slows the data access. This situation could potentially be fixed by either assigning the EBS node to a larger instance or by striping the access to the data over multiple nodes.

To understand why Copy&Compute outperforms Stream&Compute, we first compared data throughput of a VM using an ephemeral disk (as in the Stream&Compute scenario) and a VM using EBS (as in the Copy&Compute scenario) by copying large data files (0.5 GB) between Hotel and Sierra using the Unix "scp" command. The results, labeled EphemeralCopy and EBSCopy, respectively, are shown in Figure 4. We found that the throughput for small instances (the ones used in our experiments) in both cases was almost the same, so that the high throughput was not correlated with using EBS. However, when we induced a 100% CPU load using the Unix "stress" tool with 8 threads spinning over sqrt ("-cpu N") and spinning over the pair (memalloc/free) ("-vm N"), we saw the throughput diminish significantly -to roughly one-fifth of the initial throughput -but again roughly equally in the EBSCopy and EphemeralCopy case. In other words, the drop in throughput is correlated to processing because not enough CPU cycles are available both to do the processing and to manage the incoming data. Since the EBS node in Copy&Compute is dedicated to just handling the network traffic and the compute nodes in Stream&Compute are using the available cores for both computation and handling the network traffic, fewer cycles are available for handling the network traffic resulting in slower transfer.

The impact of CPU on I/O has been demonstrated before; in HPC systems [START_REF] Dorier | Damaris: How to Efficiently Leverage Multicore Parallelism to Achieve Scalable, Jitter-free I/O[END_REF], dedicating a core for handling the data has been shown to improve the overall performance and in [START_REF] Feller | On the Performance and Energy Efficiency of Hadoop Deployment Models[END_REF], the authors report a significant impact on data throughput due to virtualization. TCP throughput degradations due to sharing the CPU between cloud VMs are discussed also in [START_REF] Gamage | Protocol responsibility offloading to improve tcp throughput in virtualized environments[END_REF]. In [START_REF] Yildirim | Network-aware end-toend data throughput optimization[END_REF], the authors identify the CPU as one of the bottlenecks that need to be overcome for data transfers.

To see how additional cores may affect this scenario, we repeated these experiments for different types of instances: Small (1 CPU), Medium (2 CPUs), and Large (4 CPUs), again shown in Figure 4. The results show that increasing the number of cores does alleviate the problem; however, since no cores are dedicated to processing data transfer (as is effectively the case in the EphemeralCopy scenario), 5 present the data rate and the compute rate for the Stream&Compute scenario under those two types of instances.

As we expected, dedicating more cores to the computation improves the overall performance. This result is consistent with the experiment presented in Figure 4. The performance, however, improves only slightly, by ∼30% for the compute rate (e.g. going from 3,400 to 4,400 events/second for 16 VMs) and by ∼25% for the data rate (e.g. going from 0.95 to 1.2 MB/s for 16 VMs). As before, since no cores are actually dedicated to processing data transfer, the threads of the CPU-intensive program are able to claim a large share of attention from the CPU.

Second, we asked the question why streaming from a remote node to application VMs (as in Stream&Compute) is more impacted by the CPU activity than streaming from a local (i.e., EBS) node (as in Copy&Compute). To answer it, we used the "netem" tool to increase latency between the EBS node and application nodes of the Copy&Compute scenario to be equivalent to the remote latency used in the Steam&Compute scenario (i.e., 84 ms). The result shows that the performances (i.e., both data and compute rate) drop to the level of the Stream&Compute scenario, with the data rate decreasing from 7.5 MB/s to 1.2 MB/s. This shows that the difference in latency is responsible for the difference in performance; we believe it is tied to the data buffering mechanism in the Atlas application.

C. Scalability for streaming

The purpose of the next set of experiment was to probe the scalability barrier of the Stream&Compute scenario: we expected that for large enough numbers of application VMs the network between the data source and the application VMs would become saturated and thus the data would start arriving at the VMs at a slower rate than it could be consumed, effectively "starving" the application and thus slowing it. For this experiment we placed the application nodes on the Hotel cloud since it has more nodes available for a larger-scale experiment. We then looked at interactions with two stream data sources: a remote source running on the Sierra cloud and a local one.

The results of the experiment are presented in Figure 6. We observe that working with the remote data source, we were unable to saturate the network for the scale we tested (up to 64 application VMs) and thus the per-VM performance remained stable. Since testbed limitations prevented us from scaling further than 64 application VMs, we instead placed the data source on the Hotel cloud (i.e. local to the application compute nodes); our intention was to see whether the higher available bandwidth between data source and application nodes would allow us to saturate the network sooner. This situation was in fact observed (Figure 6) when scaling to 64 VMs: the average compute and data rates drop by about half as the aggregate throughput at which the application VMs can receive exceeds the data sources ability to send. We hypothesized that this can be fixed by streaming from multiple data sources, and this proved to be the case, as can be again seen in Figure 6. We therefore conclude that it is possible to scale beyond this bottleneck by adjusting the number of data sources with respect to the aggregated data rates of the compute VMs. Similar results in non-cloud scenarios were also shown by [START_REF] Khanna | A dynamic scheduling approach for coordinated wide-area data transfers using gridftp[END_REF]. We further probed to see how much improvement can be obtained by adding data sources. In our next experiment we kept the number of application compute VMs fixed at 64 and increased the number of data sources. The results are shown in Figure 7. After an initial significant increase in performance when moving from 1 to 2 data sources, the performance gains are smaller when going beyond 2 data sources. This upper bound is due to the physical infrastructure, as we reach the limit of the cluster capabilities (i.e., network switches and Ethernet links). We recall that the nodes have 1 Gigabit Ethernet link, which means that the average aggregated throughput to the data sources, of about 100 MB/s to each, almost consumes it completely.

D. Choosing the VM Instance Types

The cloud cost model for cloud providers such as Amazon EC2 or Microsoft Azure links compute power of instances to their price. For example, Microsoft Azure starts from s base price charged for a Small instance with 1 CPU and doubles it as the instances become more powerful (Medium = 2x, Large=4x, xLarge=8x), consistently with the resources provided (CPUs, memory, ephemeral storage). Thus, buying 1 Extra-Large instance is intended to be roughly equivalent to buying 2 Large, 4 Medium, or 8 Small instances. We therefore investigated next whether this is indeed the case in the context of the Copy&Compute scenario.

The interesting question is whether it makes any difference from the performance point of view to choose one type of instances or another in a setup such as the Copy&Compute scenario. Figure 8 presents the results corresponding to the Copy&Compute performance obtained when using 8 Small, 4 Medium, 2 Large, or 1 xLarge OpenStack VMs. In each case, the computation consists of 7 processes computing the events and 1 process acquiring data from a remote data source, potentially executing within a different VM. The VMs are run in a multitenancy environment. The results show that the data rates for all four cases are similar but slightly better for Small VMs, indicating that scheduling between VMs on the same hypervisor provides better isolation for computation and communication than does scheduling functions within one VM. The computation rates improve steadily for smaller VMs, with a 20% perfor-mance improvement for the Small versus xLarge instances. The reason is performance degradation due to the memory and disk interference that appear between processes.

This observation indicates that, given the existing balance of inter-and intra-hypervisor scheduling policies, it is better to take, within the same budget, smaller VMs rather than bigger ones. Additionally, such a strategy can potentially harvest more physical resources, since the VMs can wind up distributed across multiple nodes.

IV. THE COST OF STREAMING

Another interesting aspect to evaluate is the cost of streaming to the cloud for the two streaming strategies. We modeled this cost using the following parameters:

• V M Cost : the hourly cost for renting a VM instance;

• N V M sComp : the number of VM instances used for computation;

• N V M sData : the number of VM instances used for data acquisition (i.e., in the Copy&Compute scenario); • CompRate T otal : the number of events processed per second in the system, for Stream&Compute (SC) or Copy&Compute (CC) scenarios; • T otalEvents: the total number of events to be processed; • Size Ev : the total size of the events to be processed; • Storage Cost : the cost of storing the data on persistent storage for the duration of the computation (i.e., in the Copy&Compute scenario); • GBH Cost : the cost of storing 1 GB of data for 1 hour, which is computed as the Storage Cost divided by the number of hours in a 30-day period (the considered number of days in a month), i.e., 720 hours. Using these parameters, we can express the cost for processing a given amount of events as the time needed to process them multiplied by the corresponding cost of the VM instances used.

T otalCost = T otalEvents CompRate T otal * (NV M sData + NV M sComp) * V MCost + StorageCost (1)
The typical cost of instances similar to those we used in our computation is $0.09 per hour for Azure Small instances (as used in our experiments) and $0.06 for Amazon VM instances. The cost model for persistent attached storage typically considers long-term storage and is quoted in gigabytes per month but actually is calculated per day. For example, Azure charges $0.07 for 1 GB per month, whereas Amazon charges $0.095 per gigabyte per month.

As an illustration, in our particular use case, we need to consider 320 million events run on 32 compute nodes over 41 minutes for the Stream&Compute scenario and 15 minutes for the Copy&Compute scenario. The latter requires one extra VM instance for the data node and 5 GB of attached persistent storage additionally leased. Overall, we reach a global cost of $1.33 for the Stream&Compute scenario compared with $0.48 for the Copy&Compute scenario, in other words, the Stream&Compute scenario is 2.77 times more expensive. Extrapolating to realistic application sizes that process billions of events for multiple hours (e.g. for 1 day) using much larger numbers of VM instances, this factor can have a substantial budget impact.

To express the price ratio between the two streaming options in the general case, we start by expending the Storage Cost from Equation 1 in terms of the price to store the events for the compute time interval: T otalEvent CompRate T otal * Size Ev * GBH Cost . Hence, by expressing the price ratio between the cost of Stream&Compute and the cost of Copy&Compute, we get the following.

T otalCostSC T otalCostCC = T otalEvents CompRate T otalSC T otalEvents CompRate T otalCC * NV M sComp * V MCost (NV M sData + NV M sComp) * V MCost + SizeEv * GBHCost (2)
We note that the GBH Cost is comparable with the V M Cost for data sizes over 720 GB. The intuition is that the price to store 1 GB of data in the public clouds (e.g., Azure or Amazon) for 1 month (i.e. 720 hours) is comparable with the price of leasing one Small VM for an hour. Hence, the amount of data to be stored per hour which matches the VM cost is 720 GB. Thus, we can approximate the hourly cost to store the data as Size Ev * V M

720

. By applying this in Equation 2 we can approximate the price ratio between the two streaming options to the following. Equation 3 can be used to approximate the price reduction obtained by the Copy&Compute strategy, which is proportional with the speedup over the Stream&Compute for the computation rate of the events. For sets of events up to 720 GB, 32 to 64 compute VMs, and up to 3-4 data nodes, the second part of the equation has values between 0.90 and 0.95. Hence, the price reduction is approximately 90%-95% times the compute speedup between the two methods. This approximation can be used also for the previously discussed cost scenario, in which the Stream&Compute scenario was shown to be 2.77 times more expensive. Doing so, we find that the price reduction is about 2.7 to 2.85 times, considering a speedup of 3, as per Section III-B.

V. RELATED WORK

As cloud computing become a viable option for executing scientific applications in areas ranging from physics [START_REF] Jackson | Seeking supernovae in the clouds: a performance study[END_REF], to chemistry [START_REF] Cala | Cloud computing for fast prediction of chemical activity[END_REF], to biology [START_REF] Hodgkinson | Parallel software architecture for experimental workflows in computational biology on clouds[END_REF] and bioinformatics [START_REF] Costan | Tomus-Blobs: Scalable Data-intensive Processing on Azure Clouds[END_REF], understanding what cloud can provide becomes increasingly interesting. Multiple studies have explored various data management strategies using existing storage options for clouds [START_REF] Yelick | The Magellan Report on Cloud Computing for Science[END_REF], [START_REF] Juve | An evaluation of the cost and performance of scientific workflows on amazon ec2[END_REF], [START_REF] Sakr | A survey of large scale data management approaches in cloud environments[END_REF], [START_REF] Edwards | High-speed storage nodes for the cloud[END_REF], [START_REF] Tudoran | Datasteward: Using dedicated compute nodes for scalable data management on public clouds[END_REF], [START_REF] Hill | Early observations on the performance of windows azure[END_REF]. These general studies focus on scientific applications that process large, unstructured sets of static input data (i.e., data that is available in the cloud storage when the processing starts and remains unchanged during the computation). Consequently, the performance aspects considered in these studies focus on I/O performance with respect to VM instance types [START_REF] Yelick | The Magellan Report on Cloud Computing for Science[END_REF], data location [START_REF] Tudoran | Datasteward: Using dedicated compute nodes for scalable data management on public clouds[END_REF], or the available storage options [START_REF] Juve | An evaluation of the cost and performance of scientific workflows on amazon ec2[END_REF], [START_REF] Hill | Early observations on the performance of windows azure[END_REF]. To the best of our knowledge, however, no previous study has considered the case of dynamic sets of independent pieces of input data (e.g., data streamed from a large network of sensors to the cloud for processing).

A number of performance evaluations concerning data analysis in the clouds focus on the MapReduce [START_REF] Dean | Mapreduce: simplified data processing on large clusters[END_REF] processing paradigm. Relevant observations for our work show the importance of an adequate ratio between compute nodes and data nodes, a ratio that depends on the type of the computation [START_REF] Feller | On the Performance and Energy Efficiency of Hadoop Deployment Models[END_REF]. Additionally, the authors report significant performance penalties due to the virtualization layer. Other work related to our study was reported in [START_REF] Fadika | Evaluating hadoop for data-intensive scientific operations[END_REF], where the authors analyzed the streaming features of Hadoop [START_REF]Hadoop[END_REF] and reported the existence of an overhead due to streaming. We have made a similar observation regarding the scalability of stream processing. Since the out-of-thebox MapReduce paradigm is not appropriate for processing continuous streams of data, variations of the paradigm were proposed to support stream-based analysis [START_REF] Alves | Flood: elastic streaming mapreduce[END_REF], [START_REF] Zaharia | Discretized streams: fault-tolerant streaming computation at scale[END_REF]. Unlike these studies, our performance evaluation does not focus on stream processing using MapReduce: instead, it considers stand-alone applications that process the stream data by using independent batch workers.

Despite the expected growth of BigData and on-line processing [START_REF] Verner | Scheduling processing of real-time data streams on heterogeneous multi-gpu systems[END_REF], most of today's analytics systems are limited to a single cluster or, more rarely, a single datacenter [START_REF]Hadoop[END_REF], [START_REF] Simmhan | Bridging the gap between desktop and the cloud for escience applications[END_REF], [START_REF] Isard | Dryad: distributed data-parallel programs from sequential building blocks[END_REF]. However, due to increasing data volumes to be processed, data starts to be distributed across multiple data centers (e.g. the data from CERN's ATLAS experiment [START_REF]ATLAS[END_REF]). Additionally, with the emergence of wide geographical-area observatories [START_REF] Baptista | Scientific exploration in the era of ocean observatories[END_REF] new compute scenarios appear in which the data sources do not reside in the same geographical location with the compute nodes. Few systems, mostly derived from the MapReduce paradigm, tackle the problem of multisite or multicluster computation [START_REF] Costan | Tomus-Blobs: Scalable Data-intensive Processing on Azure Clouds[END_REF], [START_REF] Wang | Mapreduce across distributed clusters for data-intensive applications[END_REF], [START_REF] Luo | Hierarchical mapreduce programming model and scheduling algorithms[END_REF]. In the case of geographically distributed stream processing little knowledge exists about the best options for data management. The most relevant work for our study is [START_REF] Kazemitabar | Geostreaming in cloud[END_REF], where the authors proposed their vision of a stream-processing system that could potentially reach high degrees of parallelism by means of a divide, conquer, and combine paradigm. Unlike that work, our study focuses on understanding how storage and compute options available on today's clouds can be best used to reach high performance under low costs when processing remote stream data.

VI. CONCLUSION

We presented and evaluated two streaming strategies in the cloud: Stream&Compute, where the data is streamed directly to the compute nodes, and Copy&Compute, where the data is first copied to the cloud and then used for computation. Our results show that, contrary to our intuition, Copy&Compute outperforms Stream&Compute by as much as three times because of the performance trade-offs present in current virtualization technology. This performance difference can have significant consequences in terms of cost, especially at higher scales.

Our evaluation highlights the impact that trade-offs in current virtualization technology can have on cloud computing scenarios. Our results also illustrate the need for hardware and virtualization technology that can provide a more controlled allocation of CPU to I/O processing.

Figure 1 .

 1 Figure 1. The two main cloud streaming scenarios: Stream&Compute (left) and Copy&Compute (right)

Figure 4 .

 4 Figure 4. Asserting the data acquisition throughput with respect to the CPU load for three different instance types

Figure 6 .Figure 7 .

 67 Figure 6.Scaling the compute VMs for local and remote stream processing. The performance metrics show the average compute rate per (left) and the data rate (right) per VM

Figure 8 .

 8 Figure 8. Comparing the performances of the compute rate (left) and data rate (right) that can be achieved for different combinations of instance types within the same budget (Small, Medium, Large, and xLarge VMs)

  NV M sData + NV M sComp + Size Ev 720[START_REF] Sakr | A survey of large scale data management approaches in cloud environments[END_REF]

ACKNOWLEDGMENT

This work was supported by Inria through the Data@Exascale Associate Team and by the U.S. Department of Energy, Office of Science, Advanced Scientific Computing Research, under Contract DE-AC02-06CH11357. This material is produced using the FutureGrid testbed supported in part by the National Science Foundation under Grant No. 0910812. Additional experiments were carried out using Windows Azure which was supported by the joint INRIA-Microsoft Research Center (Z-CloudFlow project). We also want to thank the Nimbus team who provided helpful information about the Nimbus and FutureGrid platforms and helped us quickly resolve the issues we encountered.


Cédric Josz

Didier Henrion

Strong duality in Lasserre's hierarchy for polynomial optimization

A polynomial optimization problem (POP) consists of minimizing a multivariate real polynomial on a semi-algebraic set K described by polynomial inequalities and equations. In its full generality it is a non-convex, multi-extremal, difficult global optimization problem. More than an decade ago, J. B. Lasserre proposed to solve POPs by a hierarchy of convex semidefinite programming (SDP) relaxations of increasing size. Each problem in the hierarchy has a primal SDP formulation (a relaxation of a moment problem) and a dual SDP formulation (a sum-of-squares representation of a polynomial Lagrangian of the POP). In this note, when the POP feasibility set K is compact, we show that there is no duality gap between each primal and dual SDP problem in Lasserre's hierarchy, provided a redundant ball constraint is added to the description of set K. Our proof uses elementary results on SDP duality, and it does not assume that K has an interior point.

Introduction

Consider the following polynomial optimization problem (POP)

inf x f (x) := α f α x α s.t. g i (x) := α g i,α x α ≥ 0, i = 1, . . . , m (1)
where we use the multi-index notation x α := x α 1 1 • • • x αn n for x ∈ R n , α ∈ N n and where the data are polynomials f, g 1 , . . . , g m ∈ R[x] so that in the above sums only a finite number of coefficients f α and g i,α are nonzero. Let K denote its feasibility set: K := {x ∈ R n : g i (x) ≥ 0, i = 1, . . . , m} 1 French transmission system operator Réseau de Transport d'Electricité (RTE), 9 rue de la Porte de Buc, BP 561, F-78000 Versailles, France.

2 INRIA Paris-Rocquencourt, BP 105, F-78153 Le Chesnay, France. cedric.josz@inria.fr 3 CNRS, LAAS, 7 avenue du colonel Roche, F-31400 Toulouse, France. henrion@laas.fr 4 Université de Toulouse, LAAS, F-31400 Toulouse, France. 5 Faculty of Electrical Engineering, Czech Technical University in Prague, Technická 2, CZ-16626 Prague, Czech Republic To solve POP (1), Lasserre [START_REF] Lasserre | Optimisation globale et théorie des moments[END_REF][START_REF] Lasserre | Global optimization with polynomials and the problem of moments[END_REF] proposed a semidefinite programming (SDP) relaxation hierarchy with guaranteed asymptotic global convergence provided an algebraic assumption holds:

Assumption 1 There exists a polynomial u ∈ R[x] such that {x ∈ R n : u(x) ≥ 0} is bounded and u = u 0 + m i=1 u i g i where polynomials u i ∈ R[x], i = 0, 1, .
. . , m are sums of squares (SOS) of other polynomials.

Nie et al. [START_REF] Nie | Optimality conditions and finite convergence of Lasserre's hierarchy[END_REF] have proven that Assumption 1 also implies generically finite convergence, that is to say that for almost every instance of POP, there exists a finite-dimensional SDP relaxation in the hierarchy whose optimal value is equal to the optimal value of the POP. Assumption 1 can be difficult to check computationally (as the degrees of the SOS multipliers can be arbitrarily large), and it is often replaced by the following slightly stronger assumption:

Assumption 2 The description of K contains a ball constraint, say g m (x) = R 2 - n i=1 x 2
i for some real number R.

Indeed, under Assumption 2, simply choose u = g m , u 1 = • • • = u m-1 = 0, and u m = 1 to conclude that Assumption 1 holds as well. In practice, it is often easy to see to it that Assumption 2 holds. In the case of a POP with a bounded feasible set, a redundant ball constraint can be added.

More generally, if the intersection of the sublevel set {x ∈ R n : f (x) ≤ f (x 0 )} with the feasible set of the POP is bounded for some feasible point x 0 , then a redundant ball constraint can also be added. As an illustration, a reviewer suggested the example of the minimization of f (x) = x 2 1 + x 2 2 -3x 1 x 2 on the unbounded set defined on R 2 by the constraint g 1 (x) = 1 -3x 1 x 2 ≥ 0. The intersection of the feasible set with the set defined by the constraint f (x) ≤ f (0) = 0 is included in the ball defined by g 2 (x) = 1x 2 1x 2 2 ≥ 0 so that the POP can be equivalently defined on the bounded set

K = {x ∈ R 2 : g 1 (x) ≥ 0, g 2 (x) ≥ 0}.
Each problem in Lasserre's hierarchy consists of a primal-dual SDP pair, called SDP relaxation, where the primal corresponds to a convex moment relaxation of the original (typically nonconvex) POP, and the dual corresponds to a SOS representation of a polynomial Lagrangian of the POP. The question arises of whether the duality gap vanishes in each SDP relaxation. This is of practical importance because numerical algorithms to solve SDP problems are guaranteed to converge only where there is a zero duality gap, and sometimes under the stronger assumption that there is a primal or/and dual SDP interior point.

In [START_REF] Schweighofer | Optimization of polynomials on compact semialgebraic sets[END_REF]Example 4.9], Schweighofer provides a two-dimensional POP with no interior point for which Assumption 1 holds, yet a duality gap exists at the first SDP relaxation:

inf x 1 x 2 s.t. x ∈ K = {x ∈ R 2 : -1 ≤ x 1 ≤ 1, x 2
2 ≤ 0}, with primal SDP value equal to zero and dual SDP value equal to minus infinity. This shows that a stronger assumption is required to ensure a zero SDP duality gap. A sufficient condition for strong duality has been given in [START_REF] Lasserre | Global optimization with polynomials and the problem of moments[END_REF]: set K should contain an interior point. However, this may be too restrictive: in the proof of Lemma 1 in [START_REF] Henrion | Inner approximations for polynomial matrix inequalities and robust stability regions[END_REF] the authors use notationally awkward arguments involving truncated moment matrices to prove the absence of SDP duality gap for a certain set K that contains no interior point. This shows that the existence of an interior point is not necessary for a zero SDP duality gap. More generally, it is not possible to assume the existence of an interior point for POPs with explicit equality constraints, and a weaker assumption for zero SDP duality gap is welcome.

Motivated by these observations, in this note we prove that under the basic Assumption 2 on the description of set K, there is no duality gap in the SDP hierarchy. Our interpretation of this result, and the main message of this contribution, is that in the context of Lasserre's hierarchy for POP, a practically relevant description of a bounded semialgebraic feasibility set must include a redundant ball constraint.

Proof of strong duality

For notational convenience, let g 0 (x) = 1 ∈ R[x] denote the unit polynomial. Define the localizing matrix

M d-d i (g i y) := γ g i,γ y α+β+γ |α|,|β|≤d-d i = |α|≤2d A i,α y α
where d i is the smallest integer greater than or equal to half the degree of g i , for i = 0, 1, . . . , m, and |α| = n i=1 α i . The Lasserre hierarchy for POP (1) consists of a primal moment SDP problem

(P d ) : inf y α f α y α s.t. y 0 = 1 M d-d i (g i y) 0, i = 0, 1, . . . , m
and a dual SOS SDP problem

(D d ) : sup z,Z z s.t. f 0 -z = m i=0 A i,0 , Z i f α = m i=0 A i,α , Z i , 0 < |α| ≤ 2d Z i 0, i = 0, 1, . . . , m, z ∈ R
Z i is n + d -d i n .
Let us define the following sets:

(x) = R 2 -n i=1 x 2 i ≥ 0 reads M k-1 (g m y) = γ g m,γ y α+β+γ |α|,|β|≤k-1 with trace equal to trace M k-1 (g m y) = |α|≤k-1 γ g m,γ y 2α+γ = |α|≤k-1 g m,0 y 2α + |γ|=1 g m,2γ y 2α+2γ = |α|≤k-1 R 2 y 2α -|γ|=1 y 2(α+γ) = |α|≤k-1 R 2 y 2α -|α|≤k-1,|γ|=1 y 2(α+γ) = R 2 ( |α|≤k-1 y 2α ) + y 0 -|α|≤k y 2α = R 2 trace M k-1 (y) + 1 -trace M k (y).
From the structure of the localizing matrix, it holds

M k-1 (g m y) 0 hence trace M k-1 (g m y) ≥ 0 and trace M k (y) ≤ 1 + R 2 trace M k-1 (y) from which we derive trace M d (y) ≤ d k=1 R 2(k-1) + R 2d trace M 0 (y) = d k=0 R 2k
since trace M 0 (y) = y 0 = 1. The 2-norm M d (y) 2 , equal to the maximum eigenvalue of M d (y), is upper bounded by trace M d (y), the sum of the eigenvalues of M d (y), which are all nonnegative. Moreover the Frobenius norm satisfies

M d (y) 2 F := M d (y) , M d (y) = |δ|≤2d A 0,δ y δ , |δ|≤2d A 0,δ y δ = |δ|≤2d A 0,δ , A 0,δ y 2 δ by orthogonality of matrices (A 0,δ ) |δ|≤2d ≥ |δ|≤2d y 2 δ because A 0,δ , A 0,δ ≥ 1
where matrices (A 0,δ ) |δ|≤2d can be written using column vectors (e α ) |α|≤d , containing only zeros apart from the value 1 at index α, via the explicit formula

A 0,δ = α + β = δ |α|, |β| ≤ d e α e T β .
The proof follows then from

|δ|≤2d y 2 δ ≤ M d (y) F ≤ n + d n M d (y) 2 ≤ n + d n d k=0 R 2k .
∀p ∈ N , 1 -1 p+1 ≤ y p 0 ≤ 1 + 1 p+1 λ min (M d-d i (g i y p )) ≥ -1 p+1 , i = 0, 1, . . . , m
where λ min denotes the minimum eigenvalue of a symmetric matrix. According to the proof of Lemma 3, for all 1 ≤ k ≤ d and all real numbers (y α ) |α|≤2d , one has

trace M k-1 (g m y) = R 2 trace M k-1 (y) + y 0 -trace M k (y).
Clearly Together with λ min (M d (y p )) ≥ -1 1+p , this yields

λ max (M d (y p )) ≤ (1 + 1 + c 1 + p ) d k=0 R 2k + c -1 1 + p
where λ max denotes the minimum eigenvalue of a symmetric matrix. Hence for all p ∈ N, the spectrum of the moment matrix M d (y p ) is lower bounded by l := -1 and upper bounded by

u := (2 + c) d k=0 R 2k + c -1. Therefore: |δ|≤2d (y p δ ) 2 ≤ M d (y p ) F ≤ √ c max(|l|, |u|)
The sequence (y p ) p∈N is hence included in a compact set. Thus there exists a subsequence which converges towards y lim such that y lim 0 = 1 and λ min (M d-d i (g i y lim )) ≥ 0, i = 0, 1, . . . , m. The limit y lim is thus included is P d , which is a contradiction. SDP problem P d is strongly infeasible which means that its dual problem D d has an improving ray [START_REF] De Klerk | Self-Dual Embeddings[END_REF]Definition 5.2.2]. To conclude that val D d = +∞, all that is left to prove is that D d = ∅. Consider the primal problem P d discarding all constraints but y 0 = 1, M d (y) 0, and M d-1 (g m y) 0. It is a feasible and bounded SDP problem owing to Lemma 3. According to Lemma 2, its dual problem must contain a feasible point (z, Z 0 , Z m ) and hence (z, Z 0 , 0, . . . , 0, Z m ) ∈ D d .

Conclusion

We prove that strong duality always holds in Lasserre's SDP hierarchy for POPs whose description of the feasible set contains a ball constraint. To preclude numerical troubles with SDP solvers, we advise to systematically add a ball constraint, combined with an appropriate scaling so that all scaled variables belong to the unit sphere. Without scaling, numerical troubles can occur as well, but they are not due to the presence of a duality gap, see [START_REF] Henrion | Detecting global optimality and extracting solutions in GloptiPoly[END_REF] and also the example of [START_REF] Waki | Strange behaviors of interior-point methods for solving semidefinite programming problems in polynomial optimization[END_REF]Section 6].

  where A 0 stands for matrix A positive semidefinite, A, B = trace AB is the inner product between two matrices. The Lasserre hierarchy is indexed by an integer d ≥ d min := max i=0,1,...,m d i . The primal-dual pair (P d , D d ) is called the SDP relaxation of order d for POP[START_REF] Henrion | Detecting global optimality and extracting solutions in GloptiPoly[END_REF]. The size of the primal variable (y α ) |α|≤2d is n + 2d n and the size of the dual variable

  , trace M k-1 (g m y) ≥ -c 1+p where c := n + d n denotes the size of the moment matrix M d (y). The following holds trace M k (y p ) ≤ R 2 trace M k-1 (y p ) + 1 + 1 + c 1 + p from which we derive trace M d (y p ) ≤ (1

  Finally, let us denote by val P d the infimum in problem P d and by val D d the supremum in problem D d . Lemma 1 int P d nonempty or int D d nonempty implies val P d = val D d . Under Assumption 2, set P d is included in the Euclidean ball of radius If P d = ∅, the result is obvious. If not, consider a feasible point (y α ) |α|≤2d ∈ P d . Let k ∈ {1, . . . , d}. In the SDP problem P k , the localizing matrix associated to the ball constraint g m

	• P * d : optimal solutions for P d ;
	• D * d : optimal solutions for D d .
	d nonempty A proof of Lemma 2 can be found in [8]. According to Lemmas 1 and 2, P * and bounded implies strong duality. This result is also mentioned without proof at the
	end of [7, Section 4.1.2].
	Lemma 3 n + d n	d k=0	R 2k
	centered at the origin.

• P d : feasible points for P d ; • D d : feasible points for D d ; • int P d : strictly feasible points for P d ; • int D d : strictly feasible points for D d ; Lemma 1 is classical in convex optimization, and it is generally called Slater's condition, see e.g. [7, Theorem 4.1.3]. Lemma 2 The two following statements are equivalent : 1. P d is nonempty and int D d is nonempty; 2. P * d is nonempty and bounded.

Proof:

  Theorem 1 Assumption 2 implies that -∞ < val P d = val D d for all d ≥ d min . Proof: Let d ≥ d min . Firstly, let us consider the case when P d is nonempty. According to Lemma 3, P d is bounded and closed, and the objective function in P d is linear, so we conclude that P * d is nonempty and bounded. According to Lemma 2, int D d is nonempty, and from Lemma 1, val P d = val D d . Secondly, let us consider the case when P d is empty. An infeasible SDP problem can be either weakly infeasible or strongly infeasible, see [10, Section 5.2] for definitions. Let us prove by contradiction that P d cannot be weakly infeasible. If P d is weakly infeasible, there must exist a sequence (y p ) p∈N such that

Acknowledgements

The authors would like to thank the anonymous reviewer for insightful comments.