Fahimeh Rahemi

Ehsan Khamespanah

Ramtin Khosravi

Improving the Performance of Actor-Based Programs Using a New Actor to Thread Association Technique

Keywords: Actors, Thread Association, Self-Adaptive Algorithm, Runtime Analysis

Finding the most efficient policy for the association of objects with threads is one of the main challenges in the deployment of concurrently executing objects, including actors. For the case of actor-based programs, libraries, frameworks, and languages provide fine tuning facilities for associating actors with threads. In practice, programmers use the default policy for the initial deployment of actors and the default policy is replaced with some other policies considering runtime behaviors of actors. Although this ad-hoc approach is widely used by programmers, it is tedious and time-consuming for large-scale applications. To reduce the time-consumption of the ad-hoc approach, a set of heuristics is proposed with the aim of balancing computations of actors across threads. This technique results in performance improvement; however, it relies on the static analysis of source codes and actors' behaviors, ends in the inappropriate configuration of systems in distributed environments. In this paper, we illustrate conditions that the proposed heuristics do not work well and propose a new approach based on the runtime profile of actors for better association of actors with threads. We also show how this approach can be extended to a fully self-adaptive approach and illustrated its applicability using a set of case studies.

Introduction

The actor model is a well-known model for the development of highly available and high-performance applications. It benefits from universal primitives of concurrent computation [START_REF] Agha | ACTORS -A Model of Concurrent Computation in Distributed Systems[END_REF], called actors. Actors are distributed, autonomous objects that interact by asynchronous message passing. This model was originally introduced by Hewitt [START_REF] Hewitt | Description and Theoretical Analysis (Using Schemata) of PLANNER: A Language for Proving Theorems and Manipulating Models in a Robot[END_REF] as an agent-based language and is later developed by Agha [START_REF] Agha | ACTORS -A Model of Concurrent Computation in Distributed Systems[END_REF] as a mathematical model of concurrent computation. Each actor provides a number of services, and other actors send messages to it to run the services. Messages are put in the mailbox of the receiver, the receiver takes a message from the mailbox and executes its corresponding service. A number of programming languages and libraries are developed for actor-based programming, e.g. Act [START_REF] Lieberman | Thinking about lots of things at once without getting confused: Parallelism in act i[END_REF] and Roset [START_REF] Woelk | Developing infosleuth agents using rosette: An actor based language[END_REF] which are discontinued and Erlang [START_REF] Armstrong | [END_REF], Salsa [START_REF] Varela | Programming dynamically reconfigurable open systems with salsa[END_REF], and Akka [START_REF]Akka[END_REF] as actively supported programming languages and libraries.

In the actor programming model, a large-scale distributed applications are developed by spawning many actors which are distributed among some computation nodes and work in parallel. Using this approach, utilizing CPUs of different nodes is crucial, needs careful mapping of actors to nodes and CPUs. Some of actor-based programming languages handle scheduling of actors on different cores on runtime, using a shared pool of threads for actors which are scheduled on CPUs by round-robin approach, including Erlang [START_REF] Francesquini | Actor scheduling for multicore hierarchical memory platforms[END_REF] and Kilim [START_REF] Srinivasan | Kilim: Isolation-typed actors for java[END_REF]. However, in the majority of the JVM-based actor languages, it is the duty of programmers to associate actors with threads, including Akka and Scala [START_REF] Haller | Actors that unify threads and events[END_REF]. This way, a programmer has to associates actors with threads using the default mapping and iteratively tune the mapping, which is a very hard job and sometimes impossible for large-scale applications.

Recently, Upadhyaya et al. in [START_REF] Upadhyaya | An automatic actors to threads mapping technique for jvm-based actor frameworks[END_REF] proposed some heuristic for the association of actors with thread. To this end, they defined an Actor Characteristics Vector (cVector) for each actor to approximate the runtime behavior of it. The details of this approach are presented in Section 2. Using cVector, actors are associated with threads using one of the predefined policies of thread-pool, pinned, and monitor policies. The main goal of this approach is to map actors to threads in a way that balances actor computational workloads and reduces communication overheads. They implemented the technique for Panini and achieved on average 50% improvement in the running times of program over default mappings [START_REF] Rajan | Capsule-oriented programming[END_REF].

Although this approach improves CPU utilization of nodes significantly, it does not take the runtime behavior of systems into account. This limitation results in inefficiencies in the performance of actor systems, particularly in cases where actors are distributed among different nodes. In this work, we address both the number of spawned actors from a specific type and the load of systems at runtime to propose a better thread association policy. To this end, we propose a new light-weight technique for capturing the runtime behavior of actors (Section 3). We show how characteristic vectors of actors have to be modified to make them appropriate for presenting runtime behaviors of actors. Also, we show how the newly proposed characteristic vector is changed during the time and thread policies of actors have to be adapted to these changes. We develop a set of case studies to illustrate the applicability of this work in Section 4.

Static Association of Actors with Threads

Actors as loosely coupled parallel entities have to be associated with threads to be allowed to pick messages from their message boxes and execute them. Dedicating one thread to each actor is the simplest approach for this purpose; however, as actor-based applications usually spawn many actors, this approach does not scale. To resolve this limitation, actor libraries provide different policies for allowing programmers to associate a shared thread with multiple actors.

Using this resolution, finding the appropriate policy for the association of a thread with a group of (or one) actors is the responsibility of programmers. Generally, three different types of policies for the actor with thread association is provided to cover the requirements of applications, called thread-pool, pinned, and monitor policies. The details of these policies are presented below.

Policies for the Association of Actors with Threads

The default and widely used policy for the thread to actor association is the thread-pool policy which uses a thread-pool with a limited number of threads for a set of actors. Usually, the number of actors is more than the number of threads and actors compete for threads. This policy efficiently works for actors which are not always busy, so the less number of threads can be shared among actors. Using thread-pool policy, there is no thread preemption while an actor is busy with executing a message and actor lose its associated thread only when finishes serving a message.

As the second alternative, using the pinned policy, an OS level thread is dedicated to an actor. This policy efficiently works for busy actors, so the overhead of frequently changing the associated thread with a pinned actor is eliminated.

Finally, the monitor policy is used for actors which perform very light activities. Using the monitor policy, the associated thread with the sender of a message is reused by the receiver actor to serve the recently sent message. When serving the message is finished, the actor gives back its associated thread to the sender of the message. Note that the associated thread with the sender actor only can be reused when both of the sender and receiver are deployed on the same node.

These three policies are provided by different actor libraries with different naming. Akka provides PinnedDispatcher, BalancingDispatcher, and CallingTh-readDispatcher to realize pinned, thread-pool, and monitor policies. Akka also provides a default dispatcher which is a realization of thread-pool policy configured with a set of general purpose values. In contrast, the scheduler of Erlang only provides thread-pool policy. Kilim as the provider of very light Java actors only provides the thread-pool policy which is implemented efficiently to be able to handle thousands of actors.

Using Characteristics Vector of Actors

As the only work which tries to propose appropriate policies for actors, Upadhyaya et al. in [START_REF] Upadhyaya | An automatic actors to threads mapping technique for jvm-based actor frameworks[END_REF] proposed a heuristic-based technique for setting policies of actors (henceforth, Static-Heuristic approach). In this approach, they defined the notation of Actor Characteristics Vector (cVector) for each actor to approximate the runtime behavior of that actor. They benefit from Actor Communication Graphs (ACG) of systems to generate cVectors. The vertices of ACG are actors of a system and there is an edge between two vertices if and only if there is a possibility of sending a message from an actor which is associated with the source vertex to the actor which is associated with the destination vertex. They also marked actors which have blocking I/O activities, actors which are computationally intensive, and actors which have many communications. As a result, cVectors of actors are created as defined below.

Definition 1 (Characteristics Vectors). Set CV as the set of the characteristics vectors of actors is defined as CV = { blk, state, par, comm, cpu | blk ∈ {true, false}∧state ∈ {true, false}∧par ∈ {low, med, high}∧comm ∈ {low, med , high} ∧ cpu ∈ {low, high}}.

2

For a given characteristic vector blk, state, par, comm, cpu for the actor ac, the interpretation of the terms is as the following:

the value of blk is true if ac represents blocking behavior, the value of state is true if at least one of the state variables of ac is accessed by more than one of its methods, the value of par is low if ac sends a synchronous message and waits for the result. It is high if ac sends an asynchronous message and does not require result. Otherwise it is med, the value of comm is low if ac does not send message to other actors. It is high if ac sends message to more than one actor. Otherwise it is med, the value of cpu is high if ac represents computational workload, i.e. having recursive call, loops with unknown bounds, or making high cost library calls.

Using this interpretation, function CV : AC → CV maps a given actor to its corresponding cVector. Here, we assumed that AC is the set of actors of a system. Note that [START_REF] Upadhyaya | An automatic actors to threads mapping technique for jvm-based actor frameworks[END_REF] does not provide a precise guideline for detecting high cost library calls and blocking behavior.

To map a cVector to a thread policy, a function is defined in Definition 2. This heuristic states that a thread has to be associated with an actor (pinned policy) that has external blocking behavior. Any other policy for these actors would lead to blocking of the executing thread and may lead to actor starvation or deadlocks. In addition, any actor that is non-blocking with high inherent parallelism, high communication, and high computation should be assigned the pinned policy. Master actors, which have the property that they delegate the work to slave actors and often wait for the result are eligible for the pinned policy.

Actors with low CPU consumption and communication do not need special attention and hence are processed by the calling actor (the actor that sends messages). Actors with other characteristic vectors can share their associated threads; so, the thread-pool policy is assigned to them.

Definition 2 (Mapping cVector to Policy

). An actor which corresponds to the cVector cv ∈ CV is mapped to a thread policy by function HF : AC × CV → {pinned, thread-pool, monitor} where:

-HF (cv) = pinned: if and only if cv is in the form of true, -, -, -, -, false, -, high, high, high , or false, -, low/med, high, low , -HF (cv) = monitor: if and only if cv is in the form of false, -, -, low/med, low , -HF (cv) = thread-pool: cv does not fit the above cases.

2

Note that in this mapping, being stateful/stateless does not matter.

Runtime Association of Actors with Threads

Although the Static-Heuristic approach for the association of actors with threads results in performance improvements, it does not consider the runtime behavior of the system. This way, both over-approximation and under-approximation of the behavior of system is inevitable and causes inefficiencies in runtime. In the following we illustrated this phenomenon and proposed a runtime approach (henceforth Adaptive-Heuristic approach) to resolve it. In addition, we showed that thread association policy is widely influenced by the deployment strategy of the application and the number of hosts of actors. So, for an efficient thread association policy, deployment strategies have to be taken into account.

Redefinition of Actors Characteristics

Performing a number of experiments, we found that two terms of cVector have to be redefined. Using the current definition, these two terms misleads heuristic in actors to threads association approaches. The first is the term that shows the level of communication among actors. As mentioned before, based on the definition of [START_REF] Upadhyaya | An automatic actors to threads mapping technique for jvm-based actor frameworks[END_REF], the value of this term in the cVector of an actor is set to High if the actor sends more than one message to other actors. However, sending messages is a very light operation which is not affected by thread policies. Instead, level of communication has to be set to High for an actor which receives many messages. Many received messages results in needs for many future computational power, which is tightly in relation with thread policies. To make this difference clear, we use the example of hub-actor in [START_REF] Upadhyaya | An automatic actors to threads mapping technique for jvm-based actor frameworks[END_REF]. Hub actors are represented by either false, , high, high, low or false, , low/med, high, high which show that they have high communication characteristics. It is because of the fact that the affinity actors (actors that hub actor communicates often) send message to the hub actor, which is in contrast with the proposed metric in [START_REF] Upadhyaya | An automatic actors to threads mapping technique for jvm-based actor frameworks[END_REF], i.e. sending many messages from a hub actor to the others result in High value for the communication level.

The other case which results in having high communication is receiving messages from actors which are developed in some other nodes. As we will show later, actors with high communication are not allowed to be mapped to the monitor policy which is essential for high-performance processing of messages which are sent from actors which are hosted by the other nodes. Note than the new definition addresses the runtime behavior of systems, so it can not be used in the approach of [START_REF] Upadhyaya | An automatic actors to threads mapping technique for jvm-based actor frameworks[END_REF].

The second term that has to be redefined is the needed computational power, addressed by CPU. The needed computation power is a runtime metric which can not be effectively estimated by static analysis. Note that this argument is valid for complex actor-based systems, since the needed computational power of simple actor models can not be estimated by having a quick look into their source codes. In the new definition, the value of CPU is related to the average consumed processor time by actor. Note that the new definition sets the needed computational power for actor types not actor instances.

In addition to modifying the definition of these two terms, we found that lifetime of actors has a significant influence in the runtime behavior of actors and has to be included in the cVectors of actors. For example, using Aggregator pattern [START_REF] Vernon | Reactive Messaging Patterns with the Actor Model: Applications and Integration in Scala and Akka[END_REF], a task is split into some very simple subtasks, delegated into newly instantiated actors. The newly instantiated actors complete their associated subtasks, send the result to the owner actor and die. Regardless of the values of the others terms of the corresponding cVector, these short-lived actors are very good candidates for being associated to the monitor policy. This way, one thread is used for performing all the simple subtasks and the overhead of releasing and reclaiming thread for doing subtasks is eliminated. Note that in this case we assumed that all of actors are deployed in the same computational node. Delegating threads using monitor policy is impossible when sender and receiver actors are deployed in different computational nodes.

Based on these changes, runtime characteristics vector (rcVector) of an actor is defined as the following. We still have no observation on the effect of being stateless/stateful, so we eliminate it in runtime characteristics vectors. Definition 3 (Runtime Characteristics Vectors). Set RCV as the set of the runtime characteristics vectors of actors is defined as RCV = { blk, par, comm, cpu, lt | blk ∈ {true, false} ∧ par ∈ {low, med, high} ∧ comm ∈ {low, med, high} ∧ cpu ∈ {low, high} ∧ lt ∈ {low, high}}.

2

The interpretation of the terms in a given rcVector blk, par, comm, cpu, lt for actor ac for the terms blk and par is the same as them the original characteristics vectors and for the other three terms is as the following:

the value of comm is low if the number of received messages per a unit of time of ac is less than this value in average case of all actors. It is high if that value is bigger than the average, and otherwise it is set to med, the value of cpu is high if the value of the needed computational time per method of ac is bigger than this value for the average case, considering all of actors. In the case of receiving messages from actors, deployed on the other computation nodes of the system, the value of cpu is set to high too. otherwise it is set to low, the value of lt is high if the lifetime of ac is bigger than the average lifetime of all of the existing actors. otherwise it is set to low.

Using this interpretation, function RCV : AC → RCV maps a given actor to its corresponding rcVector. To map a rcVector to a thread policy, a function is defined as below.

Thread Pool

Towards a Self-Adaptive Approach

Using runtime mapping algorithm improves performance of systems but there is an open question on how the actors must be configured at their instantiation point. It is clear that before running systems communication level, CPU consumption, and lifetime of actors are unknown, so finding the appropriate mapping is impossible for almost all of the actors (except for actors with blocking behaviors). Therefore, a default thread policy must be assumed for all of the actors and it must be changed during the execution of the system. This adaptation is crucial for making the runtime approach possible. To this aim, we propose the adaptation algorithm which is presented in Figure 1. Actors are initially use thread-pool policy and change their thread policy upon detecting any permanent changes in the values of communication level, CPU consumption, and lifetime of their rcVectors. The labels of arrows in Figure 1 shows that which changes trigger that possible adaptation. For example, "CPU +" label on arrow between thread-pool and pinned shows that for actors which thread-pool policy increasing the value of CPU results in changing the policy to pinned. Performing this adaptation, after some amount of time the system meets its high-performance steady state.

In addition to resolving the initial mapping of actors to thread policies, the adaptation policy helps in resolving inefficiencies, caused by changes in the load profile of systems (e.g. changes in the number of clients, the operational servers, etc.). Runtime changes in the load profile of a system my change the characteristics of an actor during the time. So, some adaptation may needed after such changes to find the new high-performance steady state. The same argument is valid for actors migration, i.e. changing host nodes of actors. Based on the proposed mapping algorithm, actors migration significantly influences association of monitor policy with actors.

Experimental Results

To illustrate the applicability of this work we prepare some case studies and show how using the Adaptive-Heuristic approach improves the performance of systems. The presented case studies are partitioned in two parts. The first part contains a number of models which are proposed in [START_REF] Srinivasan | Kilim: Isolation-typed actors for java[END_REF]. The second part contains an example which shows runtime changes in load profile and the number of actors. We illustrate how the new approach adapts policies based on the encountered changes.

Models Without Runtime Adaptation

We use some of the models proposed in [START_REF] Srinivasan | Kilim: Isolation-typed actors for java[END_REF] and develop a simulator for pure actor programs. For the design of the simulator we consider both multi-node and multi-processor environments. This way, a number of threads are spread among nodes and each node schedules its own threads using its associated processors. Using this simulator, the models are developed without need for dealing with the complexities of the real-world Java actor programming. In addition, having simulator, we run models in different infrastructure configurations and monitor pure impact of thread association policies to the runtime execution of models.

In the following, we present an intuitive description and deployment diagram for each model. We also present a figure which compares the termination time of the model for three cases of using default thread-pool policy, the Static-Heuristic approach, and the Adaptive-Heuristic approach. The best approach has the smallest termination time, as it consumes the provided computation power better that the others.

Request Dispatcher. We develop RequestDispatcher example, i.e. message routing among a set of senders and receivers. This model contains three different actors which are Sender, Receiver, and Dispatcher. Sender actors pass messages to the Receiver actors via Dispatcher. The actor model of RequestDispatcher is shown in Figure 2.

As presented in [START_REF] Upadhyaya | An automatic actors to threads mapping technique for jvm-based actor frameworks[END_REF], based on the characteristics vector of the actors, the Static-Heuristic approach maps Sender and Dispatcher actors to the thread-pool policy, and Receiver to the monitor policy. This mapping only works for single node deployment of actors as upon deploying Dispatcher and Receiver in different nodes, there is no way for sharing Dispatcher threads with receivers. In addition, heavy weighted receivers may block dispatchers and reduce the performance of the system. The Adaptive-Heuristic approach proposes changing the policy of Dispatcher to the pinned policy and the policy of Receiver to thread-pool. Dispatcher as the bottleneck of the model, has to be able available permanently; so, a thread has to be associated with it. Also, in the case of deploying Receivers and Dispatcher in different nodes, there is no need for changing the policy of receivers, as they do not reuse the thread which is associated with Dispatcher.

Changing the number of senders and receivers resulted in the following figure for the completion time of the model. Two Level Hadoop Yarn Scheduler. Hadoop is a framework for MapReduce, a programming model for generating and processing large data sets [START_REF] Dean | Mapreduce: simplified data processing on large clusters[END_REF]. MapReduce has undergone a complete overhaul in its latest release, called MapReduce 2.0 (MRv2) or YARN [START_REF] White | Hadoop -The Definitive Guide: Storage and Analysis at Internet Scale[END_REF]. The fundamental idea of YARN is to split up the major functionalities of the framework into two modules, a global Resource Manager and per-application Application Master. On a Hadoop cluster, there is a single resource manager and for every job there is a single application master.

In this example, we modeled a pipeline of two instances of MapReduce clusters, depicted in Figure 4. Based on the characteristics vector of the actors, the Static-Heuristic approach maps ResourceManager actors to the pinned policy, and ApplicationMaster actors are mapped to the monitor policy. However, the work load of the second ResourceManager and ApplicationMasters are shaped by the first ResourceManager and ApplicationMaster. The Adaptive-Heuristic approach proposes pinned policy at the starting point of the first ResourceManager and changes it to thread-pool in some configurations. Based on the light weight load of the second ResourceManger, the adaptive policy proposes monitor policy for this actor. Comparison among completion times of the model in different configurations is depicted in Figure 5. File Search. Document indexing and searching model [START_REF] Upadhyaya | An automatic actors to threads mapping technique for jvm-based actor frameworks[END_REF] is the third case study that we developed. This model contains four different actors which are FileCrawler, FileScanner, Indexer, and Searcher. FileCrawler periodically visits directories which their paths are given at the start point and sends a message to FileScanner upon finding a newly modified file. To increase the verity of the number of actors in this model, we used only one crawler actor. FileScanner processes the given file and asks one of the available Indexers to index the file. Indexer performs hash-based indexing and stores the extracted information. The Searcher actor serves the search request which are sent by users. The actor model of FileSearch is shown in Figure 6(a As presented in [START_REF] Upadhyaya | An automatic actors to threads mapping technique for jvm-based actor frameworks[END_REF], the Static-Heuristic approach maps FileCrawler and Searcher actor to the pinned policy, Indexer to the monitor policy, and FileScanner to the thread-pool policy, based on the characteristics vector of the actors. The same as the previous example, this mapping only works for single node deployment of actors. The Adaptive-Heuristic approach proposes changing the policy of Indexer to the pinned policy. Also, in case of deploying FileCrawler in the node which contains FileScanner, it proposes changing the policy of FileCrawler to the thread-pool policy, as there is no need for associating one thread for its periodic behavior. Experiments showed that there is a very light improvements in using the new approach.

Bang Model. The last model we developed is the Bang benchmark which simulates many-to-one message passing. As shown in Figure 7(a), in this model there is one receiver and multiple senders that flood the receiver with messages. Based on the CVector of actor, the receiver actor is mapped to the monitor policy and senders are mapped to the tread-pool policy, using the Static-Heuristic approach.

The results of [START_REF] Upadhyaya | An automatic actors to threads mapping technique for jvm-based actor frameworks[END_REF] shows that the Static-Heuristic approach improves the performance of the system in comparison to the default policy but it does not provide the best mapping. Assume that these actors are deployed as shown in Figure 7(b). In this configuration, mapping receiver to monitor does not result in reusing the threads of senders as the actors are deployed in two different machines. In this case, the receiver actor has to be mapped to the pinned policy to be able to process the requests upon receiving their corresponding messages, as made by the new approach. However, experiments showed that there is no difference between using the Static-Heuristic approach and the adaptive-Heuristic one (based on the deployment of Figure 7(b)) as, there is no thread interference between the senders and the receiver.

A Model With Runtime Adaptation

We presented an example in the second part of experimental result which is the model of a FilmService system, shown in Figure 8(a). In this example, clients want to stream a movie from film servers. A client spawns a FilmRequest actor to search for the movie in servers. The FilmRequest actor sends messages to all of the servers and the first server which can provide the movie, spawns a Connection actor to start streaming. Besides, there are some indexer actors which are responsible for indexing the movies in the servers to make searching for movies easier.

In contrast to the aforementioned models, the load profile of actors in the FileService model may change during the time. This change takes place by requests migration when a server crashes. As soon as detecting a crashed server, requests which are sent to that server are distributed among other servers and the status of the crashed server is changed to repairing. Servers will back to service after passing a repairing period. The crash times of servers are generated by a Poisson distribution and we make sure that there is no case where all of the servers are in repairing state.

Preparing the cVector of actors of FileService for the Static-Heuristic approach results in mapping all actors to thread-pool policy, except the Client actor. However, the efficient mapping of the Server actors deeply depends on the load profile of the system. Assume that the actors are deployed as shown in Figure 8(b). In this configuration, having many film requests needs mapping Server actors to pinned to be able to process the request. This mapping reduces the performance of Indexer of that node but increase the performance of the system in general. By reducing the number of request, the mapping has to be changed to thread-pool to allow Indexer use more CPUs.

To illustrate the applicability of the Adaptive-Heuristic approach, we simulated the model in different configurations. In this case, instead of computing the average completion time of tasks, we simulated the model for a long period of time and measure the utilization of CPUs using the Adaptive-Heuristic, Static-Heuristic, and default policy approaches. This way, as the needed computation power of all tasks are the same for the three approaches, the best policy has to fully utilize CPUs. So, better utilization of CPUs means completing more tasks in a given period of time.

As shown in Figure 9, the Adaptive-Heuristic approach is the only case that shows an acceptable behavior when the number of processors is increased. The figure also shows that the inefficiency of the Static-Adaptive approach is increased by increasing the number of CPUs; however, the Adaptive-Heuristic approach encounters a very small performance penalty.

We also examined the behavior of the model in the presence of many servers, depicted in 10. This figure shows that increasing the number of servers results in a light decrease in the performance of the system in the case of using the Static-Heuristic approach which is in contrast with the light increase which is depicted in the case of the default policy and the Adaptive-Heuristic approach.

Conclusion, Discussion and Future Works

In this paper, we proposed a new approach for associating threads to actors of a system. Applying the previously proposed approaches results in performance improvement; however, it relies on static analysis of source codes and actors' behaviors. In practice, relying on the static analysis of the codes and ignoring the runtime load profile of the application results in inappropriate configuration of systems in distributed environments. In contrast, the self-adaptive approach tunes the mapping of the actors based on the captured information during the execution of the system. In this approach, the needed information can be gathered using very light-weight processes. Comparing the new approach with the old one using a set of case studies showed that the self-adaptive approach improves the performance of the systems in most cases.

Although we showed that the proposed approach results in performance improvements, as the results are computed using an actor simulation engine they may change in the real deployment of the actor models. So, we planned to develop the adaptation engine in Akka as the future work of this work. We also planned to develop more examples to show the effectiveness of the approach in different configurations.

Fig. 1 .

 1 Fig. 1. An overview of the thread policy adaptation algorithm

2

 2

Fig. 2 .Fig. 3 .

 23 Fig. 2. The RequestDispatcher example

Fig. 4 .

 4 Fig. 4. The MapReduce example

Fig. 5 .

 5 Fig. 5. The completion time of the Hadoop Yarn Scheduler model in different configurations

Fig. 6 .

 6 Fig. 6. The FileSearch example

Fig. 7 .

 7 Fig. 7. The Bang example

Fig. 8 .

 8 Fig. 8. The FilmService example

Fig. 9 .

 9 Fig. 9. Comparing CPU utilization in the case of having different number of CPUs

  )

	Searcher	File Scanner	Searcher	File Scanner		File Crawler
	File Crawler	Indexer 1 Indexer n Indexer 1	Indexer 1	Indexer 2	. . .	Indexer n

Acknowledgments

The work on this paper has been supported in part by the project "Self-Adaptive Actors: SEADA" (nr. 163205-051) of the Icelandic Research Fund.


Christopher Eibel

Christian Gulden

Wolfgang Schröder-Preikschat

Tobias Distler

Strome: Energy-Aware Data-Stream Processing

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

Introduction

Distributed data-stream-processing systems such as Twitter's Heron [START_REF] Kulkarni | Twitter Heron: Stream processing at scale[END_REF] or Spark Streaming [START_REF] Zaharia | Discretized streams: Faulttolerant streaming computation at scale[END_REF] handle millions of inputs per day, resulting in massive computations that require large amounts of energy. The purpose of the computations is multi-faceted and depends on the provided services (e.g., machine learning [START_REF] Biem | IBM Infosphere Streams for scalable, real-time, intelligent transportation services[END_REF], graph computation [START_REF] Iyer | Time-evolving graph processing at scale[END_REF], geo streaming [START_REF] Kazemitabar | Geostreaming in cloud[END_REF]). With inputs in many cases being related to user actions, the workload of a data-stream-processing system usually varies over time, often following diurnal patterns that are characteristic for data-center applications [START_REF] Atikoglu | Workload analysis of a largescale key-value store[END_REF][START_REF] Decandia | Dynamo: Amazon's highly available key-value store[END_REF]. As a result, such a system in practice does not constantly need to provide peak performance but instead is able to save energy during periods of low and medium workloads. For this purpose, data-streamprocessing systems typically rely on techniques at different levels: First, they offer the possibility to dynamically reconfigure the number of servers in the system depending on the workload that currently needs to be processed [START_REF] Cardellini | Elastic stateful stream processing in Storm[END_REF][START_REF] Cerviño | Adaptive provisioning of stream processing systems in the cloud[END_REF][START_REF] Li | Enabling elastic stream processing in shared clusters[END_REF]. Second, on each server, the systems exploit power-saving techniques such as dynamic voltage and frequency scaling (DVFS) [START_REF] Horowitz | Low-power digital design[END_REF][START_REF] Semeraro | Energy-efficient processor design using multiple clock domains with dynamic voltage and frequency scaling[END_REF] to increase the energy efficiency of each server individually. In this paper, we focus on the latter problem, identify drawbacks of DVFS in the context of data-stream-processing systems, and present an approach to further improve a server's energy efficiency.

When DVFS is activated on a server, the server's operating system monitors CPU utilization and dynamically regulates processor speed, for example, decreasing the processor frequency when utilization is low in order to reduce power demand. Our experiments with Heron confirm this strategy to be effective, but also show that using DVFS a server often still requires significantly more power than would actually be necessary to handle the current workload. As the main reason for this behavior we identified the fact that due to only taking CPU utilization into account, DVFS needs to make pessimistic assumptions on its own interference with the application. Consequently, in an effort to avoid performance degradation, an operating system applying DVFS not always configures the hardware to be in its most energy-efficient power state.

Building on our findings, we developed an approach, Strome, that allows a data-stream-processing system to leverage the so far unused energy-saving potential. In contrast to DVFS, Strome's decision-making process is not limited to system-level information obtained on the local server, but instead combines application-level performance metrics collected in the entire system. As a key advantage, this allows Strome to precisely assess its impact on application performance and to coordinate energy-saving mechanisms across different servers. To minimize the power demand of a server, Strome relies on modern hardware features such as RAPL [START_REF]Intel 64 and IA-32 architectures software developer's manual[END_REF] that enable system software to specify upper limits for the power demand of components (e.g., CPU, memory), which are then enforced by the hardware. Compared with DVFS, RAPL offers the benefit of taking effect at a wider range of power configurations, thereby especially enabling energy savings at low and medium workload levels.

With Strome being implemented as a part of the data-stream-processing platform, the applications running on top of it can profit from our approach without requiring any modifications. Our prototype implementation based on Heron shows that Strome seamlessly integrates with real-world systems. Furthermore, our experimental evaluation with varying workloads confirms that Strome is able to dynamically adjust power-demand limits in an effective and coordinated fashion, thereby automatically adapting to workload changes. Altogether, this paper makes the following contributions:

1. It shows how the lack of application awareness significantly reduces the amount of energy the state-of-the-art DVFS-based approach is able to save. 2. It presents the Strome approach of making data-stream-processing systems energy aware and applying energy-saving techniques in a way that takes their effects on application performance into account. 3. It evaluates the effectiveness of Strome at the example of Heron for a variety of common data-stream-processing applications.

The remainder of the paper is structured as follows. Section 2 summarizes our analysis of the effects of DVFS on Heron and consequently uses our findings to motivate the Strome approach. Section 3 presents details on the Strome design and implementation. Section 4 evaluates our Strome prototype, Section 5 discusses the adaptability, portability, and scalability of the Strome approach. Finally, Section 6 summarizes related work and Section 7 concludes.

Background and Problem Analysis

Data-stream-processing applications are implemented using topologies, that is, compositions of processing elements and operators, which each handle different parts of the overall task. Figure 1 shows an example of such an application that performs analyses on the browsing behavior of Web-page visitors. For this application, two source nodes ("spouts") provide the input data, which is then routed through a series of worker nodes ("bolts") responsible for preprocessing and analyzing the user data. In a last step, a joiner bolt combines the produced results and forwards them to the data sink. To execute a topology, data-streamprocessing systems such as Heron [START_REF] Kulkarni | Twitter Heron: Stream processing at scale[END_REF] distribute the topology's nodes across the available servers and manage the data flow between machines. For this purpose, the systems typically rely on a dedicated master process that, amongst other things, determines node placement, starts the topology, and collects application performance metrics (e.g., throughput and latency) during execution.

As the amount of input that needs to be processed usually varies throughout the course of a day, servers in a data-stream-processing system do not always have to provide peak performance, but instead are able to save energy during periods of reduced workloads. The technique applied in the vast majority of today's servers to achieve this is dynamic voltage and frequency scaling (DVFS) [START_REF] Horowitz | Low-power digital design[END_REF][START_REF] Semeraro | Energy-efficient processor design using multiple clock domains with dynamic voltage and frequency scaling[END_REF]. DVFS allows the operating system of a server to dynamically adjust a CPU's frequency (and voltage) depending on current system load, thereby adapting the CPU's performance capabilities to the work that needs to be performed.

To analyze the impact of this approach on the power demand of a datastream-processing system, we conduct an experiment with the user-behavioranalysis application presented above. As shown in Figure 2, during the experiment we vary the workload in order to examine the effectiveness of DVFS at different levels. For comparison, we evaluate three different power-configuration modes (i.e., DVFS, performance, and power-save), which are implemented by different power governors [START_REF]Linux CPUFreq[END_REF]. Only for DVFS, the operating system actually varies voltage and frequency based on the current load level. In contrast, the two other modes work with constant configurations, always operating the CPU at its maximum (performance mode) and minimum (power-save mode) frequency, respectively. Consequently, the results of these power-configuration modes can serve as baselines for the highest and lowest possible power demand. Performance DVFS Power-Save Fig. 2. Throughput and power-demand comparison at different workload levels for a Web-user-behavior-analysis application using three power-configuration modes.

Our measurement results in Figure 2 show that (1) the power demand of the overall system to a large extent depends on the work performed by CPUs and that (2) DVFS is able to support the full spectrum of workloads: For high workloads, it allows the application to achieve maximum performance at the cost of an increased power demand, while for low and medium workloads DVFS effectively reduces power demand. However, the results of this experiment also reveal that DVFS does not necessarily apply the most power-efficient configuration in all cases. At a throughput of 20 kTuples/s, for example, the power-save configuration achieves a 12 % (64 W vs. 73 W) lower power demand than DVFS, despite processing the same workload. This effect is a consequence of the problem that DVFS tries to minimize its interference with the application while using CPU utilization as the only metric to estimate current performance requirements. To compensate the lack of knowledge, DVFS needs to make pessimistic assumptions on its own negative impact on application performance, which for some load levels results in processor configurations with non-optimal power demand.

One way to minimize power demand would be to always execute a system in power-save mode, however, our measurement results illustrate that this also is not an option as it usually prevents an application from processing high workloads. For the example application, the maximum throughput achievable in power-save mode is 21.5 kTuples/s, that is, less than half of the 50 kTuples/s maximum throughput for DVFS and the performance mode. This shows that there is a tradeoff between saving as much energy as possible for a particular workload and being able to handle arbitrary workloads.

To overcome this issue, we developed Strome, an approach that enables data-stream-processing systems to achieve additional energy savings for low and medium workloads without sacrificing the ability to support high workloads. To minimize energy demand, Strome sets CPU power-demand limits and dynamically adapts them in the face of varying workloads. In contrast to DVFS, Strome does not consider CPU utilization but instead directly takes application performance metrics into account. Furthermore, Strome is not focused on a single server but addresses saving energy as a distributed problem, thereby better reflecting the distributed nature of today's data-stream-processing systems.

Strome

In this section, we first give an overview of the Strome approach (Section 3.1) and then provide details on the particular mechanisms it uses to minimize energy demand (Section 3.2), to coordinate power-saving techniques across servers (Section 3.3), and to adapt to varying workloads (Section 3.4).

Overview

Strome's main goal is to minimize the energy demand of a data-stream-processing system without decreasing application performance. To achieve this, depending on the current workload Strome dynamically adjusts individual limits for the amount of power each server in the system is allowed to use, thereby saving energy at low and medium workloads while still being able to exploit all available processing resources at high workloads.

As shown in Figure 3, Strome seamlessly integrates with existing datastream-processing systems such as Heron. On a system's master server, Strome introduces an additional module, the global regulator, which is responsible for determining the current workload level and consequently also decides whether reconfigurations of power-demand limits are necessary. Once the global regulator chooses to modify the power-demand limit of a worker server, it instructs a dedicated Strome module on this server, the local regulator, to implement the new power cap. To determine the power-cap values to apply, Strome's global regulator relies on a metrics database that is populated by the local regulators and contains server-specific information on power demand and application performance. Relying on this database, the global regulator, for example, detects if a server is in danger of becoming overloaded and, as a countermeasure, can increase the affected server's power-demand limit.

Being integrated with the data-stream-processing system, Strome does not require any application-specific configuration or modification. When faced with an unknown workload, the global regulator initiates an autonomous and coordinated mechanism to learn and apply the most suitable power-demand limits for all worker servers in the system, and afterwards stores this knowledge for future use. Using the same mechanism and information, Strome is also able to dynamically adjust power caps in order to react to workload changes.

Performance-aware Power Capping

Strome saves energy during periods of low and medium workloads by reducing the amount of power a server is allowed to use. To enforce power-demand limits, we exploit the fact that modern servers are equipped with power-management features such as running average power limit (RAPL) [START_REF]Intel 64 and IA-32 architectures software developer's manual[END_REF]. RAPL offers finegrained control over a machine's maximum power demand, taking a specific power-demand value in watt (→ power-cap value) as input. In contrast to DVFS, RAPL does not only change frequencies and voltages but uses additional hardware features such as throttling the CPU's clock, which enables further energy reductions and allows RAPL to strictly adhere to the requested power cap.

To minimize its impact on the application, Strome applies power caps in a performance-aware manner: it constantly monitors application performance in order to assess the effects of newly set power caps. In this context, two application-performance metrics are of particular interest:

-Throughput: The number of data tuples that are processed by the system per second. Strome's main goal is to maximize energy savings while ensuring that this metric is not affected by the selected power caps. -Back-pressure activity: The amount of time input data is buffered due to a processing element being overloaded. This metric is crucial because it serves as an early indicator that a current power limit may be set too low, allowing Strome to quickly detect the need for a reconfiguration.

To collect these performance metrics for a topology, Strome utilizes the built-in metrics facility already available in data-stream-processing systems. Heron, for example, provides this service by running a separate metrics-manager component on each worker node that registers and forwards node-specific runtime statistics (e.g., tuples processed per bolt on that node) to the master's metrics database (cf. Figure 3). That is, the metrics database contains, for example, concrete throughput values (i.e., tuples/s) for each machine in the cluster, ordered by a timestamp and categorized by the type of processing element (spout, bolt, sink). Based on this information, Strome is able to calculate the topology's total throughput by adding up the throughput values of all sinks.

Apart from performance metrics, Strome's worker servers also retrieve machine-specific power values and forward them to the master's metrics database.

To obtain such values, we exploit the fact that RAPL can not only be used to implement power caps but also to measure the power demand of CPUs and other hardware components (e.g., DRAM, GPU, memory controller). As the values provided by RAPL only reflect a part of a server's overall power demand, in addition, we also feed the metrics database with results from an external measuring device that cover the entire power demand of all worker servers in the system, including the power demand of hardware components such as disks, RAMs, mainboards, fans, and power-supply units. Combining this knowledge about power demand with the application-performance metrics offered by the data-stream-processing system, Strome has all the information necessary to minimize the power caps for each server depending on the current workload.

Coordinated Distributed Power Capping

Controlling the power-demand limits of all worker servers a topology is running on, Strome is able to coordinate the selection and implementation of power caps across servers. For this purpose, the control logic in Strome's global regulator runs in a continuous feedback loop, which involves multiple steps it periodically executes. Its goal is to determine the best power cap for each machine to save the largest possible amount of energy while not interfering with application performance. The specific power-cap values depend on the type of topology, the number and types of worker servers in the cluster, and the current throughput.

At startup, the global regulator retrieves information on the topology from the data-stream-processing system's master, which in particular includes the set of worker machines that participate in executing the topology. In a second step, the global regulator resets all power caps on all servers (i.e., it makes sure that all machines have no power-demand limit), meaning that the servers run with maximum performance possible. Knowing the servers that participate, the global regulator can also obtain their current individual power-demand values from the metrics database. Utilizing this knowledge, the global regulator then sorts the servers in decreasing order of their current power-demand values and thereby defines the order in which the individual power cap for each server will be determined. The rationale behind this approach is to start with the worker server that contributes the most to the system's overall power demand and is therefore likely to offer the highest savings in absolute numbers.

Starting with the first worker server on the list, the global regulator repeats the steps sketched in Figure 4 for each server in the system. Initially, it instructs the server's local regulator to set the power cap to the current power demand of the server (Line 2). Next, the global regulator gradually decreases the power cap (Line 4) until a processing element in the topology is no longer able to handle its inputs (Line 6), which is detected based on the back-pressure time information provided by the metrics database (see Section 3.2). At this point, the global regulator resets the server's power cap to the last value known to support the current workload (Line 7) and advances to the next server. This process continues until power caps for all worker servers have been determined. Relying on back-pressure information as an indicator to decide when a suitable power-demand limit for a server is reached has two main advantages: First, the buffering time of processing elements usually quickly increases in case of overload situations and therefore allows the global regulator to revoke a low power cap before it can have a broader impact on overall application performance. Second, in contrast to throughput, for example, the back-pressure metric enables the global regulator to distinguish between a performance decrease that is caused by a low power cap and a performance decrease that is the result of fewer input data flowing into the system. Only in the former case, the global regulator needs to abort its efforts for the current server, while in the latter case, it is able to continue by further reducing the server's power-demand limit.

Dynamic Adaptation to Varying Workloads

In addition to power capping all servers in a coordinated manner, Strome's global regulator also continuously monitors application performance and periodically reevaluates the current configuration in order to be able to dynamically react to workload changes. To speed up the adaptation process, the global regulator maintains a power-cap database containing the power-cap values previously determined for different throughputs using the mechanism discussed in Section 3.3. As a key benefit, this database allows the global regulator to quickly adjust power-demand limits for known workload levels.

The power-cap database is implemented as a map that stores throughput categories (e.g., 10 kTuples/s, 20 kTuples/s, 30 kTuples/s, etc.) as keys and the corresponding sets of power caps as values, together with metadata such as the identifiers of worker servers. If a periodic reevaluation is due and an entry matches the current workload, the global regulator immediately instructs the local regulators to apply the power caps from the database. Otherwise, the global regulator initiates the distributed power-capping mechanism described in Section 3.3, which will eventually lead to the creation of a new database entry. The same also happens for entries whose throughput values are already in the database in case the set of worker servers in the system changes.

When creating the power-cap database, the global regulator does not aim at collecting entries that reflect equidistant throughput categories. Instead, the regulator targets a finer granularity of database entries for low and medium workloads, as these are the ranges that offer the highest energy savings. This approach allows Strome to improve its effectiveness while limiting the costs necessary for populating and maintaining the power-cap database.

Evaluation

In this section, we evaluate Strome with multiple Heron applications to determine the amount of power that can be saved for different topologies in the presence of a varying throughput (i.e., the number of incoming tuples to process each second). Furthermore, we show the potential of applying power caps and evaluate Strome's ability to dynamically adapt to changing throughput levels.

Experimental Environment

We conduct our experiments on a cluster of three homogeneous worker nodes, comprising servers with an Intel Xeon E3-1275 v5 processor (Skylake architecture, 8 cores with Hyper-Threading, SpeedStep, and Turbo Boost enabled, 3.40 GHz). All machines are connected via switched 1 Gbps. To gather nodespecific energy and power values, we use RAPL. In addition, for a complete view of the whole cluster in terms of power and energy demand, including all the machines' hardware, we use an external, high-precision measuring device, the Microchip MCP39F511 that provides results with a measuring error of only 0.1 % [START_REF][END_REF]. We our Strome prototype based on Heron version 0.14.6 and run all machines on Ubuntu version 16.04.3 LTS. For comparison, we repeat the experiments with standard Heron while DVFS (i.e., the Linux ondemand governor) is enabled on all machines. As topologies, we use typical data-streamprocessing applications that are either CPU or memory bound. The evaluated applications, including short descriptions, are summarized in Table 1.

Table 1. Overview of the evaluated Heron applications.

Application Description

ClickAnalysis

Analyzing origins and interactions of users (IP addresses) with Web pages to gain insights into their Web-browsing behavior.

BargainIndex

A financial benchmark that calculates the volume-weighted average price by adding all shares multiplied by their share price and dividing the resulting value by the total number of shares in a specific period.

WordCount

Splits sentences into words and counts each word's occurrences.

TweetAnalysis

Spam and sentiment detection on a stream of incoming tweets.

Topology-Dependent Power-Capping Efficiency

In our first experiment, we are interested in the effectiveness of applying power caps to different topologies. For this purpose, we start all workers with the topologies in Table 1 and apply power caps for different throughput values. We compare these values to the results obtained with the same setup of workers, topologies, and inputs when relying on DVFS. All power-demand values reflect the total power demand of all processing workers, measured with the MCP39F511. Figure 5 presents the resulting power values for all topologies within their individual throughput range. Each data point in the graph represents the lowest power value achievable for the respective throughput when each worker is capped with a specific power value. Depending on the topology type, the maximum throughput varies between 4 and 300 kTuples/s, which illustrates the diverse characteristics of the four topologies evaluated. Comparing the power demand of Strome with DVFS shows that Strome provides better power efficiency for all topologies. The amount of power that can be saved varies with the topology. For example, while the maximum power savings for ClickAnalysis is almost 17 W at a throughput of 45 kTuples/s (i.e., Strome's power demand is 15 % lower compared with DVFS), the power savings achievable for BargainIndex at a throughput of 90 kTuples/s are as high as 35 W (25 %).

Table 2 contains a selection of the chosen power caps for all workers for ClickAnalysis and WordCount and specific throughput levels. The power-cap values show that there is no linear relationship between throughput and power caps to set. Moreover, the ratio between power-cap values of two worker servers is not always the same over the whole throughput range. For example, with WordCount, at a throughput of 280 kTuples/s, the caps for Worker 2 and Worker 3 are equally set to 6.875 W, whereas at 140 kTuples/s, the cap of Worker 2 can be set 2.5 W higher than the cap of Worker 3 .

From these results we conclude that it is important to not only know the effects of a power cap based on a single throughput value, but to investigate the whole throughput range. This also means that Strome's approach of determining the power caps at runtime is favorable over a pure model-based approach, which is hard and cumbersome to establish for all kinds of topologies.

Coordinated, Distributed Power-Capping Analysis

In our second experiment, we investigate the coordinated, distributed powercapping mechanism presented in Section 3.3. In this scenario, the ClickAnalysis topology is executed with a constant throughput of 40 kTuples/s. The global regulator runs with an empty power-cap database; that is, it does not know beforehand which caps are favorable for the currently applied throughput. Figure 6 shows the complete throughput trend over the entire experiment (top) and the RAPL power-demand values measured for each worker servers (bottom).

The global regulator's capping procedure begins to cap all workers one by one, starting with Worker 1 since it has the highest power demand. After a few seconds, the power demand of Worker 1 is reduced by about 5 W and an additional 1 W at 20 s into the experiment. As the global regulator detects overload, it immediately raises Worker 1 's power cap again so that no throughput deviation is observable. Next, the power caps for Worker 2 and then for Worker 3 are decreased in the same way and finally set to approximately 15 W and 5 W, respectively. Thus, the dynamic power-demand amount, which is the power demand that does not depend on fans, disks, peripherals etc., is reduced by almost 12 %.

Finding a suitable power cap for each worker may induce a small temporary throughput reduction (e.g., at 120 s into the experiment when finding the power cap for Worker 3 ), which is acceptable considering the energy savings possible with Strome over the long period of time a data-stream-processing application is typically running. Overall, based on this experiment we can conclude that even in phases where no power-cap database entry has been established yet, the effects on performance are minimal, showing the practicability of our approach. Fig. 6. Analysis of Strome's coordinated, distributed power-capping mechanism in terms of throughput and individual per-worker power demand for ClickAnalysis.

Dynamic-Adaptation Analysis

In our third and final experiment, we rely on the BargainIndex topology to evaluate how Strome behaves in the presence of a varying throughput when its power-cap database is already filled with entries for a topology. For comparison, we repeat the experiment with DVFS using the same varying workload. Figure 7 presents the obtained measurement results for throughput (top), the total power demand of all involved workers (middle), and the end-to-end latency (bottom). During the execution of the topology, Strome periodically queries its database for power caps that are suitable for the current throughput. This process happens immediately; that is, in contrast to the experiment in Section 4.3 no extra time is necessary to determine the power caps to set, allowing Strome to instantly save power compared with DVFS. The values for end-to-end latency show that Strome's adaptation process and the powercapping measure only have a small effect on the time it takes to process input data. Although the execution times within the processing elements (e.g., inside a bolt) may increase with lower power caps, other factors (e.g., communication) have significantly more influence so that these execution times are negligible.

In summary, this experiment confirms that, independent of the current workload, Strome is able to provide the necessary throughput performance without deteriorating the end-to-end latency of the data-stream-processing application.

Discussion

This section discusses adaptability, portability, and scalability aspects of the Strome approach and its implementation. Adaptability. Throughout this paper, we have shown the concept and energy-demand improvements of Strome at the example of Twitter Heron. However, the Strome approach is not limited to Heron only. Other systems such as Apache Spark Streaming [START_REF]Apache Spark Streaming[END_REF] and Apache Storm [START_REF]Apache Storm[END_REF] can also benefit from Strome without requiring major changes to their implementation. is a result of our design choice to keep large parts of the local-and global-regulator logic independent of the actual data-stream-processing system running. Moreover, most data-stream-processing systems share design principles, such as the type and behavior of processing elements (e.g., join) or the performance metrics of interest (e.g., throughput), allowing Strome, for example, to use a metrics database with a similar schema (i.e., mapping throughput to power caps) for different data-stream-processing systems. Access to application metrics is inherently important for Strome to make better power-management decisions than DVFS. Apache Storm and Apache Spark Streaming, for instance, both have a built-in metrics API that pushes data via different metric reporters to a central database.

Portability. Strome uses power-capping techniques that are available on a large set of today's machines. All new Intel processors, even the mobile versions, are equipped with this feature. Older AMD processors support the application power management (APM) [START_REF]Advanced Micro Devices, Inc. BIOS and kernel developer's guide (BKDG) for AMD family 15h models 30h-3Fh processors[END_REF] feature, which is comparable to RAPL, while newer ones provide an interface that is more similar to the original RAPL interface from Intel [START_REF][END_REF]. Therefore, a multitude of systems in modern data centers offer a feature equal or similar to RAPL. For the few servers for which this is not the case, it is possible to access commonly available features such as P-states to adjust the frequency-voltage pair directly. This way, Strome can adapt these pairs based on its knowledge about the application and still reduce the machines' power and energy demand significantly compared with standard DVFS.

Scalability. Strome's algorithm for determining the individual power caps for all servers sequentially finds the power cap for each server involved in the execution of a topology. While this mechanism is sufficient for the number of machines used in our evaluation setting, it might be too time intensive for a larger set of machines. For such cases we suggest the following two refinements: First, if the workload level changes before the algorithm completes, the energy regulator can save the power caps learned so far and resume the algorithm for the remaining servers when the throughput returns to its original level at a later point in time. Second, if there are many machines in the cluster, we suggest to group them by certain criteria (e.g., the type of work they have to accomplish) and apply the same power cap to all machines in the group at once. If the energy regulator detects an overload situation, it resets the power caps of all machines in the group to the last known value that supports the current workload (as is the case for the algorithm that sequentially caps single machines, cf. Section 3.3). In consecutive runs, it is then possible to further refine the found cap by splitting up the original group and reevaluating the cap for the smaller sub groups.

Related Work

Strome is related to several research domains. We present and discuss the most relevant to its design principles in the following.

Data-Center Power Management. There are multiple approaches at different system levels which tackle the problem of reducing energy demand. DVFS is a technique that is applied on all kinds of machines, whereas power capping is typical for large data centers, where the power demand is limited because of thermal reasons (equipment protection) [START_REF] Azimi | Fast decentralized power capping for server clusters[END_REF], not tripping circuit breakers [START_REF] Wu | Dynamo: Facebook's data center-wide power management system[END_REF], or to stay within a provisioned budget [START_REF] Bhattacharya | The need for speed and stability in data center power capping[END_REF]. Hardware-enforced power capping to reduce power demand has previously been investigated for different types of applications [START_REF] Eibel | Towards energy-proportional state-machine replication[END_REF][START_REF] Eibel | Empya: Saving energy in the face of varying workloads[END_REF][START_REF] Lo | Towards energy proportionality for large-scale latency-critical workloads[END_REF][START_REF] Rountree | Beyond DVFS: A first look at performance under a hardware-enforced power bound[END_REF]. Strome is the first work to use this technique in order to improve energy efficiency in data-stream-processing systems.

Elastic Data-Stream Processing. Elastic data-stream processing commonly refers to data-stream processing systems that are dynamically reconfigured at runtime. Li et al. [START_REF] Li | Enabling elastic stream processing in shared clusters[END_REF] present Stela for the data-stream-processing system Apache Storm. Generally, this framework aims at increasing throughput while decreasing the average request response times, without respecting energy or power demand. The system is monitored to dynamically adapt the number of workers or the parallelism of operators in the topology. Cardellini et al. [START_REF] Cardellini | Elastic stateful stream processing in Storm[END_REF] propose a similar approach where operators may also be relocated. StreamCloud, introduced by Gulisano et al. [START_REF] Gulisano | Stream-Cloud: An elastic and scalable data streaming system[END_REF], lowers the distribution-algorithm overhead for queries and thus improves the general scalability of the data-stream-processing system. Cerviño et al. [START_REF] Cerviño | Adaptive provisioning of stream processing systems in the cloud[END_REF] engage scaling by allocating or de-allocating virtual machines with regard to the input rate. In contrast to Strome, relocation has the effect of considerably impacting latency and should therefore be handled carefully. These mentioned approaches are orthogonal to Strome as they directly change the data-stream-processing system itself (including the API), whereas Strome is adaptable to different systems without requiring deep modifications.

Quality-of-Service Awareness. Besides saving energy, Strome ensures quality of service, an important goal in a wide range of different fields. For example, Zhu et al. [START_REF] Zhu | Event-based scheduling for energy-efficient QoS (eQoS) in mobile Web applications[END_REF] propose energy-efficient quality-of-service awareness for mobile Web applications. Heinze et al. [START_REF] Heinze | Online parameter optimization for elastic data stream processing[END_REF] propose another approach for data-streamprocessing systems where quality of service is balanced with monetary costs. Again, energy awareness is not respected in this approach. De Matteis et al. [START_REF] Matteis | Keep calm and react with foresight: Strategies for low-latency and energy-efficient elastic data stream processing[END_REF] do consider energy demand in addition to also optimizing for latency in datastream-processing systems; however, the presented framework relies on DVFS to reduce energy demand, which we have shown in this work does not exploit the full energy-saving potential of today's hardware. Dhalion by Floratuou et al. [START_REF] Floratou | Dhalion: Self-regulating stream processing in Heron[END_REF], like Strome, is based on Twitter Heron. Dhalion is a self-regulating system that does not care about energy or power at all. Using Dhalion, it is not required to restart and reload the whole topology when changing participating hardware components. Thus, we see great potential in extending our work to incorporate Dhalion for using not only available power-capping features such as RAPL but also to dynamically switch between diverse hardware components that each have their strengths in terms of energy demand for certain throughput regions.

Resource-Aware Multiprocessor Systems Due to the parallel nature of topologies, data-stream-processing systems greatly benefit from being executed on machines with multiple processors. Apart from challenges such as programmability, adaptivity, scalability, physical constraints, reliability, and fault tolerance [START_REF] Teich | Invasive computing: An overview[END_REF], resource and energy awareness are crucial problems in current and future many-core systems. These issues motivate the new computing paradigm invasive computing [START_REF] Teich | Invasive algorithms and architectures[END_REF], which introduces resource-aware programming. The invasive-computing paradigm gives applications the possibility to distribute their workloads depending on the availability and status of the underlying system resources. A key resource here is energy. Systems on a chip consisting of hundreds or thousands of cores are concerned with inherent power limitation just to avoid overheating or even blowing of circuitry (dark silicon [START_REF] Esmaeilzadeh | Dark silicon and the end of multicore scaling[END_REF]). Strome shares the insight that extensive knowledge about the structure and progress of applications (i.e., data-stream-processing topologies) is greatly useful for an energyefficient operation of computing systems. Whether this knowledge is obtained statically, for example at design-exploration time, as it is the case with invasive computing, or dynamically at runtime, is not decisive for Strome itself.

Conclusion

Applying DVFS reduces the energy demand of data-stream-processing systems, but for low and medium workloads in many cases cannot exploit the full energysaving potential. To address this problem we have presented Strome, a poweraware technique that relies on power capping to save energy in data-streamprocessing systems without affecting performance. Compared to DVFS, Strome operates at the granularity of multiple servers which enables the technique to select and implement power caps in a coordinated fashion across servers. For its reconfiguration decisions, Strome does not rely on CPU utilization, but instead explicitly takes application performance (e.g., throughput) and overload metrics (e.g., the buffering times of processing elements) into account and is therefore able to minimize its own interference with the application. Our evaluation with different application scenarios on common server hardware has shown very good power-saving results compared to the widely used traditional DVFS. Furthermore, our experiments have confirmed Strome's ability to adapt to varying workloads. As future work, we plan to investigate the Strome approach in the context of other data-stream-processing systems. In addition, we want to explore further power-saving techniques such as using heterogeneous workers.

3 MasterFig. 1 .

 31 Fig. 1. Example of the logical topology of a data-stream application for analyzing Web-user behavior (left) and its physical distribution among three servers (right).

Fig. 3 .

 3 Fig. 3. Overview of Strome's basic system architecture at the example of Heron.

1 void 3 while(true): 4 server 6 ifFig. 4 .

 13464 Fig. 4. Basic Strome algorithm for determining the individual power cap for a server.

Fig. 5 .

 5 Fig. 5. Power-capping efficiency for the evaluated Heron topologies.

Fig. 7 .

 7 Fig. 7. Throughput, power-demand, and end-to-end-latency evaluation of DVFS versus Strome with a fully established power-cap database for BargainIndex.

Table 2 .

 2 A selection of power-cap values (where Capx corresponds to the power cap set on Workerx) for WordCount as well as ClickAnalysis and specific throughput values.

	Application	Throughput [Tuples/s]	Cap1 [W]	Cap2 [W]	Cap3 [W]
		70,000	2.5	6.875	5
	WordCount	140,000 210,000	2.5 3.125	11.25 22.5	8.125 22.5
		280,000	4.375	28.75	28.75
		10,000	3.5	3.5	5
	ClickAnalysis	20,000 30,000	6.875 11.875	6.875 9.375	3.75 3.75
		40,000	21.25	15.625

Acknowledgments

We thank Heiko Janker for his help towards a precise and accurate total-power measuring device and the anonymous reviewers for their insightful feedback. This work was partially supported by the German Research Council (DFG) under grant no. DI 2097/1-2 ("REFIT") and grant no. SFB/TR 89 ("InvasIC").