Cosmin Arad
email: cosmin@sics.se

Jim Dowling
email: jdowling@sics.se

Seif Haridi

Message-Passing Concurrency for Scalable, Stateful, Recongurable Middleware

Keywords: Passing Concurrency for Scalable, Stateful, Reconfigurable Middleware component model, message-passing, compositional concurrency, dynamic reconguration, multi-core execution, reproducible simulation, distributed systems architecture

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

Introduction

In recent times, there has been a marked increase in the use of programming languages and frameworks that support message-passing concurrency (MPC) to build high performance servers [START_REF] Chen-Becker | The Denitive Guide to Lift: A Scala-based Web Framework[END_REF][START_REF] Anderson | CouchDB: The Denitive Guide Time to Relax. 1st edn[END_REF]. The main reasons for the renewed interest in MPC are that it scales well on multi-core hardware architectures and that it provides a simple and compositional concurrent programming model, free from the quirks and idiosyncrasies of locks and threads. Another reason is that high performance non-blocking sockets map easily to MPC applications. In addition to this, functional programming implementations of MPC, such as Erlang [START_REF] Armstrong | Making reliable distributed systems in the presence of software errors[END_REF] and Scala actors [START_REF] Haller | Scala actors: Unifying thread-based and event-based programming[END_REF], have the benet of being suitable for building middleware that is dynamically recongurable. Due to its stateless nature and support for message passing, Erlang supports the construction of software that can be safely upgraded online. Message processing can be temporarily suspended in modules marked for upgrade, and the problem of transferring state from the old module to the new module is largely avoided.

The challenge we address in this paper is how to provide support for both MPC and dynamic reconguration in a framework for building high-performance middleware that lends itself more naturally to a stateful programming model, supporting session and application state. Existing stateful programming models and frameworks that support dynamic reconguration, such as component frameworks [START_REF] Coulson | A generic component model for building systems software[END_REF], are not designed for MPC support and they do not decouple their execution model from component code. As a result, they cannot run the same code in both simulation and production environments.

In previous work on dynamically recongurable middleware, component models, such as OpenCom [START_REF] Coulson | A generic component model for building systems software[END_REF] and Fractal [START_REF] Bruneton | The Fractal component model and its support in Java: Experiences with auto-adaptive and recongurable systems[END_REF], developed mechanisms such as explicit dependency management, component quiescence, and recongurable connectors for safely adapting systems online. However, the style of component interaction, based on blocking interface invocation, precludes compositional concurrency in these models making them unsuited to present day multi-core architectures.

Our work is also relevant within the context of popular non-blocking socket frameworks that are used to build high performance event-driven server applications [START_REF] Welsh | Seda: an architecture for well-conditioned, scalable internet services[END_REF], such as Lift [START_REF] Chen-Becker | The Denitive Guide to Lift: A Scala-based Web Framework[END_REF] and Twitter's Finagle [START_REF] Wampler | Scala web frameworks: Looking beyond lift[END_REF] for Scala, and Facebook's Tornado [START_REF] Dory | Introduction to Tornado[END_REF] for Python. Kompics' asynchronous event programming framework allows it to seamlessly integrate dierent non-blocking networking frameworks (such as Netty, Apache Mina, and Grizzly) 3 as pluggable components.

Kompics is a message-passing, concurrent, and hierarchical component model with support for dynamic reconguration. The broad goal of Kompics is to raise the level of abstraction in programming distributed systems. We provide constructs, mechanisms, architectural patterns, as well as programming, concurrency, and execution models that enable programmers to construct and compose reusable and modular distributed abstractions. We believe this is an important contribution because it lowers the cost and accelerates the development and evaluation of more reliable distributed systems. The other main motivation for our asynchronous event programming framework is performance, particularly for high-concurrency networked applications.

Through a case-study of a scalable key-value store, we show that the performance of traditional event-driven programming does not have to come at the cost of more complex programs. Using encapsulation, components can hide event-driven control ow and support component reuse. We leverage encapsulation when testing Kompics systems, by enabling the same component code to be run in both simulation and production systems. To support the easy specication of simulation experiments, we introduce a domain-specic language that provides constructs for generating simulation experiment scenarios containing thousands of nodes.

3 http://www.jboss.org/netty; http://mina.apache.org; http://grizzly.java.net.

A summary of our key principles in the design of Kompics are as follows. First, we tackle the increasing complexity of modern distributed systems through hierarchical abstraction. Second, we decouple components from each other to enable dynamic system evolution and runtime dependency injection. Third, we decouple component code from its executor to enable dierent execution environments.

Component Model

Kompics is a component model targeted at building distributed systems by composing protocols programmed as event-driven components. Kompics components are reactive state machines that execute concurrently and communicate by passing data-carrying typed events, through typed bidirectional ports, connected by channels. This section introduces the conceptual entities of our component model and its programming constructs, its execution model, as well as constructs enabling dynamic reconguration, component life-cycle and fault management.

Concepts in Kompics

The fundamental Kompics entities are events, ports, components, event handlers, subscriptions, and channels. We introduce them here and show examples of their denitions with snippets of Java code. The Kompics component model is programming language independent, however, we use Java to illustrate a formal denition of its concepts.

Events. Events are passive and immutable typed objects having any number of typed attributes. The type of an attribute can be any valid type in the host programming language. New event types can be dened by sub-classing old ones.

Here are two example event type denitions in Java 4 :

 class Message extends Event {  Address source;  Address destination;  }  class DataMessage extends Message {  Data data;  int sequenceNumber;  }
In our Java implementation of Kompics, all event types are descendants of a root type, Event. We write DataMessage⊆Message to denote that DataMessage is a subtype of Message. In diagrams, we represent an event using the Event graphical notation, where Event is the event's type, e.g., Message.

Ports. Ports are bidirectional event-based component interfaces. A port is a gate through which a component communicates with other components in its environment by sending and receiving events. A port allows a specic set of event types to pass in each direction. We label the two directions of a port as positive (+) and negative (-). The type of a port species the set of event types that can traverse the port in the positive direction and the set of event types that can traverse the port in the negative direction. Concretely, a port type denition consists of two sets of event types: a positive set and a negative set. There is no sub-typing relationship for port types.

Here are two example port type denitions in Java 5 :

4 We omit the constructors, getters, setters, access modiers, and import statements for clarity. 5 The code block in the inner braces represents an instance initializer. The positive and negative methods populate the respective sets of event types. In our implementation, a port type is a (singleton) object (for fast dynamic event ltering). In this example we dene a Network port type which allows events of type Message (or a subtype thereof ) to pass in both (`+' and `-') directions. The Timer port type allows ScheduleTimeout and CancelTimeout events to pass in the `-' direction and Timeout events to pass in the `+' direction.

 class Network extends

Conceptually, a port type can be seen as a service or protocol abstraction with an event-based interface. It accepts request events and delivers indication or response events. By convention, we associate requests with the `-' direction and responses or indications with the `+' direction. In our example, a Timer abstraction accepts ScheduleTimeout requests and delivers Timeout indications.

A Network abstraction accepts Message events at a sending node (source) and delivers Message events at a receiving node (destination) in a distributed system.

A component that implements a protocol or service will provide a port of the type that represents the implemented abstraction. Through this provided port, the component will receive the request events and trigger the indication events specied by the port's type. In other words, for a provided port, the `-' direction is incoming to the component and the `+' direction is outgoing from the component.

In Figure 1, the MyNetwork component provides a Network port and the MyTimer component provides a Timer port. In diagrams, we represent a port using the Port + -graphical notation, where Port is the type of the port, e.g., Network.

We represent components using the Component notation.

When a component uses a lower level abstraction in its implementation, it will require a port of the type that represents the abstraction. Through a required port, a component sends out the request events and receives the indication/response events specied by the port's type, i.e., for required ports, the `-' direction is outgoing from the component and the `+' direction is incoming to the component. quired Network port of the FailureDetector. This allows, e.g., Message events sent by the FailureDetector to be received by MyNetwork.

Channels forward events in both directions in FIFO order. In diagrams, we represent channels using the channel graphical notation. We omit the channel name when it is not relevant.

Handlers. An event handler is a rst-class procedure of a component. A handler accepts events of a particular type (and subtypes thereof ) and it is executed reactively when the component receives such events. During its execution, a handler may trigger new events and mutate the component's local state. The handlers of one component instance are mutually exclusive, i.e., they are executed sequentially. This alleviates the need for synchronization between dierent event handlers of the same component accessing the component's mutable state, which greatly simplies their programming.

Here is an example event handler denition in Java:

 Handler<Message> handleMsg = new Handler<Message>() {  public void handle(Message message) {  messages++; // ← component-local state update 
System.out.println("Received from " + message.source);  }};

In diagrams, we use the hEvent graphical notation to represent an event handler, where h is the handler's name and Event is the type of events accepted by the handler, e.g., Message.

Subscriptions. A subscription binds an event handler to one component port, enabling the handler to handle events that arrive at the component on that port.

A subscription is allowed only if the handler's accepted event type, E, is allowed to pass by the port's type denition. In other words, E must be one of (or a subtype of one of ) the event types allowed by the port's type denition to pass in the direction of the handler.

Figure 3 illustrates the handleMsg handler from our previous example being subscribed to a port. In diagrams, we represent a subscription using the graphical notation.

In this example, the subscription of handleMsg to the Network port is allowed because Message is in the positive set of Network; handleMsg will handle all events of type Message or a subtype of Message, received on this Network port.

Components. Components are event-driven state machines that execute concurrently and communicate asynchronously by message-passing. In the host programming language, components are objects consisting of any number of local MyComponent Network +  handleMsg Message Fig. 3. The handleMsg event handler is subscribed to the required Network port of MyComponent. As a result, handleMsg will be executed whenever MyComponent receives a Message event on this port, taking the event as an argument. state variables and event handlers. Components are modules that export and import event-based interfaces, i.e., provided and required ports. Each component is instantiated from a component denition.

Here is an example component denition in Java:

 class MyComponent extends ComponentDefinition {  Positive<Network> network = requires(Network.class); // ← required port  int messages; // ← local state  public MyComponent() { // ← component constructor  System.out.println("MyComponent created.");  messages = 0;  subscribe(handleMsg, network);  }  Handler<Message> handleMsg = new Handler<Message>() { ... };  }
In this example we see the component denition of MyComponent which was illustrated in Figure 3. Line 2 species that the component has a required Network port. The requires method returns a reference to a required port, network, which is used in the constructor to subscribe the handleMsg handler to this port (line 7). The type of the required port is Positive Network because, for required ports the positive direction is incoming into the component. Both a component's ports and event-handlers are rst-class entities which allows for their dynamic manipulation.

Components can encapsulate subcomponents to hide details, reuse functionality, and manage system complexity. Composite components enable the control and dynamic reconguration of entire component ensembles as if they were single components. Composite components form a containment hierarchy rooted at a Main component (see Figure 4). Main is the rst component created when the runtime system starts and it recursively creates all other sub-components. Since there exist no components outside of Main, Main has no ports.

Here is the Main component specication in Java:

 class Main extends ComponentDefinition {  Component net,

Kompics Operations

While presenting the Kompics concepts we have already introduced some of the basic operations on these concepts: subscribe, create, and connect. These have counterparts that undo their actions: unsubscribe, destroy, and disconnect, and these have the expected semantics. Here is the code for destroy and disconnect using our previous example: A fundamental command in Kompics is trigger, which is used to (asynchronously) send an event through a port. In the next example, MyComponent handles a MyMessage event due to its subscription to its required Network port.


Upon handling the rst message, MyComponent triggers a MyMessage reply on its Network port and then it unsubscribes its handleMyMsg handler, thus handling no further messages. Figure 5 illustrates MyComponent. In diagrams, we denote that an event handler may trigger an event on some port, using the Event graphical notation.

Publish-Subscribe Event Dissemination

Components are unaware of other components in their environment. A component can communicate, i.e., handle received events and trigger events, only through the ports visible within its scope. The ports visible in a component's scope are its own ports and the ports of its immediate sub-components. Ports and channels forward triggered events toward other connected components, as long as the types of events triggered are allowed to pass by the respective port type specications. Hence, component interaction is dictated by the connections between components as congured by their enclosing parent component.

Component communication follows a message-passing publish-subscribe model.

An event triggered (published) on a port is forwarded to other components by all channels connected to the other side of the port (Figure 6). As an optimization, our runtime system avoids forwarding events on channels that would not lead to any compatible subscribed handlers. An event received on a port is handled by all compatible handlers subscribed to that port (Figure 7). An Init event is guaranteed to be the rst event handled. When a component subscribes an Init event handler to its Control port in its constructor, the component will not handle any other event before a corresponding Init event. Start and Stop events allow a component (which handles them) to take some actions when the component is activated or passivated. A component is created passive. In the passive state, a component can receive events but it will not execute them. (Received events are stored in a port queue.) When activated, a component will enter the active state (executing any enqueued events). Handling life-cycle events is optional for a component.

 class MyComponent extends

To activate a component, a Start event is triggered on its control port, and to passivate it, a Stop event is triggered on its control port. Here is an example snippet of code possibly executed by a parent of myComponent:

MyComponent Control +  startH Start stopH Stop initH MyInit Start Stop Init Fault +  Control +  MyInit  Init

Fault Management

Kompics enforces a fault isolation and management mechanism inspired by Erlang [START_REF] Armstrong | Making reliable distributed systems in the presence of software errors[END_REF]. A software fault or exception thrown and not caught within an event handler is caught by the runtime system, wrapped into a Fault event and triggered on the Control port, as shown in Figure 9.

Dynamic Reconguration

Kompics enables the dynamic reconguration of the component architecture without dropping any of the triggered events. In addition to the ability to dynamically create and destroy components, connect and disconnect ports, subscribe and unsubscribe handlers, Kompics supports four channel commands to enable safe dynamic reconguration: hold, resume, plug, and unplug. The hold command puts the channel on hold. The channel stops forwarding events and starts queuing them in both directions. The resume command has the opposite eect, resuming the channel. When a channel resumes, it rst forwards all enqueued events, in both directions, and then keeps forwarding events as ususal.

The unplug command, unplugs one end of a channel from the port where it is connected, and the plug command plugs back the unconnected end to a (possibly dierent) port.

To replace a component c1 with a new component c2 (with similar ports), c1's parent, p, puts on hold and unplugs all channels connected to c1's ports; then, p passivates c1, creates c2 and plugs the unplugged channels into the respective ports of c2 and resumes them; c2 is initialized with the state dumped by c1 and activated. Finally, p destroys c1.

Implementation

We have implemented Kompics in Java. In this section we discuss some of the implementation details related to the runtime system, component scheduling, We implemented a rich library of components and ports that provide basic distributed systems abstractions. For example, we have three dierent implementations for the Network abstraction using Apache MINA, Netty, and the Grizzly network library, respectively. Each of these components implements automatic connection management, message serialization, and Zlib compression. The choice of implementations is congurable -for example, CATS in section 4 uses Grizzly with Kyro for message serialization.

dierent
Multi-core Component Scheduling. The Kompics execution model admits an implementation with one lightweight thread per component. However, as Java has only heavyweight threads, we use a pool of worker threads for concurrently executing components. Every component is marked as idle (if it has no events awaiting execution), ready (if it has one or more events waiting in ports to be executed in handlers), or busy (if an event is currently being executed in a handler). Each worker has a dedicated queue of ready components. Workers process one event in one component at a time and one component cannot be processed by multiple workers at the same time. Thus, the Kompics execution model guarantees that handlers of a single component instance execute mutually exclusively.

Workers may run out of ready components to execute, in which case they engage in work stealing [START_REF] Blumofe | Scheduling multithreaded computations by work stealing[END_REF]. Work stealing involves a thief, a worker with no ready components contacting a victim, the worker with the highest number of ready components, and stealing a batch of half of its ready components. Stolen components are moved from the victim's work queue to the thief 's work queue.

From our experiments, batching shows a considerable performance improvement over stealing small numbers of ready components. To improve concurrency, the work queues are implemented as lock-free queues, meaning that the victims and thieves can concurrently consume ready components from their queues.

By designing components as reactive state machines and scheduling them using a pool of worker threads, we provide a simple programming model that leverages multi-core machines without any extra programming eort.

Deterministic Simulation Mode. We provide a special scheduler for reproducible system simulation. The system code is executed in deterministic simulation provided it does not attempt to create threads. In simulation mode, the system's bytecode (including any binary libraries) is instrumented to intercept all calls for the current time and return the simulated time. Therefore, without editing any of its source code, the system can be executed deterministically in simulated time. Library code for secure random number generators is also instrumented to use the same seed and achieve determinism. Attempts to create threads are also intercepted and the simulation halts since it would not be able to guarantee deterministic execution.

Testing and Programming in the Large. Kompics supports test-driven development through both unit-testing and integration-testing. Firstly, since components are implemented in Java classes, a component can be mocked, so that the individual handlers can be unit-tested. Secondly, integration tests (tests covering more than one component) can be implemented as Java unit tests running the tested subsystem in simulation mode, enabling systems to be built and validated using standard continuous integration platforms. To this end, we used Apache Maven to organize the structure and manage the artifacts of the Kompics component library. The complete framework counts more than 100 modules. We organize the various Kompics concepts into abstraction and component This approach also enables deploy-time composition.

Case Study: A Scalable, Consistent Key-Value Store

To put into perspective the Kompics concepts, patterns, and dierent execution modes, we present a case study of a key-value store called CATS that provides a simple API to get and put key-value pairs, while guaranteeing linearizable consistency in partially synchronous, lossy, partitionable and dynamic networks [START_REF] Arad | CATS: Linearizability and partition tolerance in scalable and self-organizing key-value stores[END_REF].

Kompics was used to develop, deploy, stress-test, and simulate CATS. This is a (non-trivial) large-scale, self-organizing distributed system with dynamic node membership. Each node in the system handles a complex mix of protocols for failure detection, topology maintenance, routing, replication, group membership, agreement, and data consistency. In the next section we highlight the component based software architecture of the system and later we show how the same system implementation designated for deployment is executed in simulation mode for debugging and testing under a wide array of concurrency and failure scenarios.

CATS Deployment Architecture

Firstly, we provide a general component framework with protocols reusable in many large-scale distributed systems (such as our key-value store or a peerto-peer (P2P) system). Such systems typically need a bootstrap procedure to assist newly arrived nodes in nding nodes already in the system in order to execute any join protocols. To this end, we have a BootstrapServer component which maintains a list of online nodes for a particular system instance. Every node embeds a BootstrapClient component which provides a Bootstrap service to the node. When a node starts, it issues a BootstrapRequest to the client which retrieves from the server a list of alive nodes and delivers a BootstrapResponse to the node. The node runs a join protocol against one or more of the returned nodes and after joining, it sends a BootstrapDone event to the client, which, from now on, will send periodic keep-alives to the server letting it know this node is still alive. The BootstrapServer evicts nodes who stop sending keep-alives.

Another reusable service provided, is a monitoring service that provides distributed tracing, where a client component at each node periodically inspects the status of various internal components, and may also log all network events for tracing. The client periodically sends reports to a monitoring server that can aggregate the status of nodes, rebuild trace trees, and present a global view of the system on a web page. The bootstrap and monitoring servers are illustrated in Figure 10 (left), within executable main components. routine since it enables us to quickly run a small-scale distributed system (without the need for remote deployment or launching of multiple processes) and we can interact with it using a web browser.

CATS Experimentation

We designed a Java domain-specic language (DSL) for expressing experiment scenarios for large-scale distributed systems. Such scenarios are interpreted by a NetworkEmulator/ExperimentDriver or NIO Framework and ExperimentDriver component. A scenario is a parallel and/or sequential composition of stochastic processes. We start each stochastic process, a nite random sequence of events, with a specied distribution of inter-arrival times.

Here is an example stochastic process: This will generate a sequence of 1000 catsJoin operations, with an interarrival time between two consecutive operations extracted from an exponential distribution with a mean of 2 seconds. The catsJoin operation is a system-specic operation with 1 parameter. In this case, the parameter is the ring identier of the joining node, extracted from an uniform distribution of [0.. We dened three stochastic processes: boot, churn, and lookups. The next code snippet shows how we can compose them into a complete experiment scenario. The experiment scenario starts with the boot process. Two seconds (simulated time) after this process terminates, the churn process starts. Three seconds after churn starts, the lookups process starts, now working in parallel with churn.

 StochasticProcess boot = new
The experiment terminates one second after all lookups are done. Putting it all together, here is how one denes and executes an experiment scenario using our Java DSL: // execution mode Note that the above code is an executable Java main-class. It creates a sce-nario1 object, sets an RNG seed, and calls the simulate method passing the simulation architecture of your system as an argument (line 14). The simulate method instruments the bytecode of the system and executes it in simulation mode, driving the simulation from the given experiment scenario. This is useful for debugging. If you want to run an interactive experiment, comment out line 14 and uncomment line 15. This will run your interactive execution architecture and drive it from the same scenario. You will be able to interact with and monitor the system over the web while the experiment is running.

 class CatsSimulationExperiment {  // system-
Discussion and Simulation Performance. We have showed the component based software architecture of a non-trivial distributed system and how the same system implementation designated for deployment can be executed in simulation mode or interactive whole-system execution. We showed how Kompics can be used to build scalable, concurrent middleware using CATS as a case study.

We also ran simulations of CATS and we were able to simulate a system of 16384 nodes in a single 64-bit JVM with a heap size of 4GB. The ratio between the real time taken to run the simulation and the simulated time was roughly 1.

For smaller system sizes we observe a much higher simulated time compression eect, as shown in Table 1.

Related Work

Kompics is related to work in several areas: concurrent programming models [1215], recongurable component models for distributed systems [START_REF] Coulson | A generic component model for building systems software[END_REF][START_REF] Bruneton | The Fractal component model and its support in Java: Experiences with auto-adaptive and recongurable systems[END_REF][START_REF] Ostrowski | Programming with live distributed objects[END_REF], Kompics's message-passing concurrency model is similar to the actor model [START_REF] Agha | Actors: a model of concurrent computation in distributed systems[END_REF], of which Erlang [START_REF] Armstrong | Programming Erlang. Pragmatic Bookshelf[END_REF], the Unix lter and pipe model, Kilim [START_REF] Srinivasan | Kilim: Isolation-typed actors for java[END_REF] and Scala [START_REF] Odersky | Scalable component abstractions[END_REF] are, perhaps, the best known examples. Similar to the actor model, message passing in Kompics involves buering events before they are handled in a rst-in rst-out (FIFO) order, thus, decoupling the thread that sends an event from the thread that handles an event. In contrast to the actor model, event buers are associated with component ports, so each component can have more than one event queue, and ports are connected using typed channels. Channels that carry typed messages between processes are also found in other message-passing systems, such as Singularity [START_REF] Fähndrich | Language support for fast and reliable message-based communication in Singularity OS[END_REF]. Connections between processes in the actor models are unidirectional and based on process-ids, while channels between ports in Kompics are bi-directional and components are oblivious to the destination of their events.

The main features of the Kompics component model, such as the ability to compose components, support for strongly-typed interfaces, and explicit dependency management using ports, are found in many existing component models, such as ArchJava [START_REF] Aldrich | Architectural reasoning in archjava[END_REF], OpenCOM [START_REF] Coulson | A generic component model for building systems software[END_REF], Fractal [START_REF] Bruneton | The Fractal component model and its support in Java: Experiences with auto-adaptive and recongurable systems[END_REF], LiveObjects [START_REF] Ostrowski | Programming with live distributed objects[END_REF], and OMN-net++ [START_REF] Varga | An overview of the OMNeT++ simulation environment[END_REF]. However, with the exception of LiveObjects, these component models are inherently client-server models, with blocking RPC interfaces.

LiveObjects has the most similar goals to Kompics of supporting encapsulation and composition of distributed protocols. Its endpoints are similar to our ports, providing bi-directional message-passing, however, endpoints in LiveObjects support only one-to-one connections. Other dierences with Kompics include: the lack of a concurrency model beyond shared-state concurrency, the lack of recongurability, and the lack of support for hierarchical components.

Although there is support for dynamic reconguration in some actor-based systems, such as Erlang, Kompics's reconguration model is based on reconguring strongly typed connections between components. Component-based systems that support similar runtime reconguration functionality use either reective techniques, such as OpenCOM [START_REF] Coulson | A generic component model for building systems software[END_REF], or dynamic software architecture models, such as Fractal [START_REF] Bruneton | The Fractal component model and its support in Java: Experiences with auto-adaptive and recongurable systems[END_REF], Rapide [START_REF] Luckham | An event-based architecture denition language[END_REF], and ArchStudio4/C2 [START_REF] Dashofy | Archstudio 4: An architecture-based meta-modeling environment[END_REF]. Kompics's reconguration model is most similar to the dynamic software architecture approaches, but a major dierence is that the software architecture in Kompics is not specied explicitly in an architecture denition language, rather it is implicitly constructed at runtime.

Other work related to Kompics are non-blocking socket frameworks that support asynchronous event programming, such as Tornado for Python [START_REF] Dory | Introduction to Tornado[END_REF] and Lift for Scala actors. Protocol composition frameworks, such as Horus [26], Appia [START_REF] Miranda | Appia, a exible protocol kernel supporting multiple coordinated channels[END_REF] and Mace [START_REF] Killian | Mace: language support for building distributed systems[END_REF], are also related, but they are specically designed for building distributed systems by layering modular protocols. Although this approach certainly simplies the task of programming distributed systems, these frameworks are often designed with a particular protocol domain in mind and this limits their generality. Mace, however, also supports the execution of the same code in both production and simulation. Finally, there are related tools for monitoring distributed systems, such as Dapper [START_REF] Sigelman | Dapper, a large-scale distributed systems tracing infrastructure[END_REF] by Google, a distributed tracing system that is built-in to a few key modules commonly linked by all applications.

In contrast, in Kompics, we have a monitoring client that execute concurrently and can be easily adapted to handle events published by any component.

Conclusions and Future Work

In this paper we presented the Kompics component model and programming framework. We showed how complex distributed systems can be built by composing simple protocols. Protocol abstractions are programmed as event-driven, message-passing concurrent components. Kompics contributes a unique combination of features well suited for the development and testing of large-scale, long-lived distributed systems, including: hierarchical component composition, dynamic reconguration, message-passing concurrency, publish-subscribe nonblocking component interaction, seamless integration of NIO frameworks, and the ability to run the same code in either production mode or reproducible simulation for testing and stepped debugging. For future work, we are investigating a Kompics front-end in Scala. This would immediately leverage the existing Java components and runtime system. Also, it has the potential for more expressive code and a succint DSL for Kompics operations.

Fig. 1 . 2 Fig. 2 .

 122 Fig. 1. The MyNetwork component has a provided Network port. MyTimer has a provided Timer port. The FailureDetector has a required Network port and a required Timer port. In diagrams, a provided port is gured on the top border, and a required port on the bottom border of a component.

Fig. 5 .

 5 Fig. 5. MyComponent handles one MyMessage event and triggers a MyMessage reply on its required Network port.  class MyComponent extends ComponentDefinition {  Positive<Network> network = requires(Network.class);  public MyComponent() { // ← component constructor  subscribe(handleMyMsg, network);  }  Handler<MyMessage> handleMyMsg = new Handler<MyMessage>(){  public void handle(MyMessage m) {  trigger(new MyMessage(m.destination, m.source), network);  unsubscribe(handleMyMsg, network); // ← reply only once  }};}

Fig. 6 .

 6 Fig. 6. When MyNetwork triggers a MessageA on its provided Network port, this event is forwarded by both channel1 and channel2 to the required Network ports of Component1 and Component2, respectively.

Figure 8

 8 illustrates the Control port type and a component that declares an Init, a Start, and a Stop handler. Typically, for each

Fig. 7 .

 7 Fig. 7. When MyNetwork triggers a MessageA event on its Network port, this event is delivered to the Network port of MyComponent and handled by both handler1 and handler2, sequentially (gured with yellow diamonds), in the order in which the two were the Network port. component denition that state initialization one denes a specic initialization event (subtype of Init) which contains component-specic conguration parameters.

Fig. 8 .

 8 Fig. 8. Every Kompics component provides a Control port by default. To this Control port, the component can subscribe Start, Stop, and Init handlers. In general, we do not illustrate the control port in component diagrams.

Fig. 9 .

 9 Fig. 9. Uncaught exceptions thrown in event handlers are caught by the runtime, wrapped in a Fault event and triggered on the control port.

  modes of execution, and component dependency management. Kompics is publicly released as an open-source project. The source code for the Java implementation of the Kompics runtime, component library, and case studies presented here, are all available online at http://kompics.sics.se. Java Runtime and Network I/O. Our Java runtime system implements the Kompics concepts and operations as well as the Kompics execution model. The Kompics runtime system supports pluggable component schedulers, decoupling component behaviour from component execution. In particular, this enables the ability to use dierent component schedulers to execute the same (unchanged) component-based system in dierent modes: parallel multi-core execution and deterministic simulation. Next subsection highlights the default scheduler.

  packages. An abstraction package contains a port together with the request and indication events of that port. A component package contains the implementation of one component with some component-specic events (typically subtypes of events dened in required ports). The source code for an abstraction or component package is organized as a Maven module and the binary code is packaged into a Maven artifact, a JAR archive annotated with meta-data about the package's version, dependencies, and pointers to web repositories from where (binary) package dependencies are automatically fetched by Maven.In general, abstraction packages have no dependencies and component packages have dependencies on abstraction packages for both the required and provided ports. This is because a component implementation will use event types dened in abstraction packages, irrespective of the fact that an abstraction is required or provided. Maven enables the reusability of protocol abstractions and component implementations. When we start a project for a new protocol implementation we just need to specify what existing abstractions our implementation depends on. They are automatically fetched and made visible in the new project.

Fig. 10 .

 10 Fig.10. Bootstrap and monitoring servers (left) exposing a user-friendly web interface for troubleshooting. Component architecture for one CATS node (right). This architecture is designated for system deployment where every CATS node executes on a dierent machine and communicates with other nodes by sending messages using Grizzly, a Java NIO non-blocking sockets framework.

class Main extends ComponentDefinition

					{
		Component net, timer, fd;	// ← subcomponents
		Channel channel1, channel2;	// ← channels
		public undo() {		//	some method
		disconnect(net.provided(Network.class), fd.required(Network.class));
		disconnect(timer.provided(Timer.class), fd.required(Timer.class));
		destroy(net);	destroy(timer);	destroy(fd);
	 }}

  2 16 ]. Here is how the catsJoin operation is dened:The catsLookup operation takes two BigInteger parameters, extracted from a (here, uniform) distribution, and generates a Lookup event that tells CATS Simulator to issue a lookup for key key at the node with identier node. As you can see above, a random node in 0..2 16 will issue a lookup for a random key in 0..2 14 . 5000 lookups are issued in total, with an exponential inter-arrival time with mean 50 milliseconds.

	 Operation2<Lookup, BigInteger, BigInteger> catsLookup
		= new Operation2<Lookup, BigInteger, BigInteger>() {
		public Lookup generate(BigInteger node,BigInteger key){
		return new Lookup(new NumericRingKey(node), new NumericRingKey(key));
	 }};
	 Operation1<Join, BigInteger> catsJoin = new Operation1<Join, BigInteger>() {
		public Join generate(BigInteger nodeKey){
		return new Join(new NumericRingKey(nodeKey));
	 }};
		It takes one BigInteger argument (extracted from a distribution) and gener-
	ates a Join event (triggered by the NetworkEmulator/ExperimentDriver on PutGet
	port). Next, we dene a churn process which will generate a sequence of 1000
	churn events (500 joins randomly interleaved with 500 failures), with an expo-
	nential inter-arrival time with a mean of 500 milliseconds.
	 StochasticProcess churn = new StochasticProcess() {{
		eventInterArrivalTime(exponential(500));//exponentially distributed, µ = 500ms
		raise(500, catsJoin, uniform(16));	//500 joins
		raise(500, catsFail, uniform(16));	//500 failures
	 }};
		Next, we dene a process to issues some Lookup events.
	 StochasticProcess lookups = new StochasticProcess() {{
		eventInterArrivalTime(normal(50, 10));//normally distributed,µ = 50ms, σ = 10ms
		raise(5000, catsLookup, uniform(16), uniform(14));
	 }};

Table 1 .

 1 Time compression eects observed when simulating the system for 4275 seconds of simulated time.

	Peers Time compression
	64	475x
	128	237.5x
	256	118.75x
	512	59.38x
	1024	28.31x
	2048	11.74x
	4096	4.96x
	8192	2.01x
	recongurable software architectures [1720], and event-based frameworks for
	distributed systems [7, 21, 22].

CATS Node Main

CATS Node

LAN latency (two message round-trips, so 4 one-way latencies), message serialization (4x), encryption (4x), decryption (4x), deserialization (4x), and Kompics runtime overheads for message dispatching and execution. In terms of scalability, for read-intensive workloads, reading 1KB values, CATS scaled on Rackspace to 96 machines providing just over 100,000 reads/sec. We refer the reader to [START_REF] Arad | CATS: Linearizability and partition tolerance in scalable and self-organizing key-value stores[END_REF] for more details on CATS performance.

CATS Simulation Architecture

We now show how we can reuse the CATS' components, without modifying their code, to execute the system in simulation mode for testing, stepped debugging, or repeatable simulation studies. This whole architecture is executed in simulation mode, i.e., using a simulation component scheduler which executes all components that have received events and when it runs out of work it passes control to the NetworkEmulator/ExperimentDriver to advance the simulation time.

Local, Interactive, Stress-Test Execution

Using the same experiment scenario used in simulation, the same system code can be executed in an interactive stress-testing execution mode. Figure 12 (right) shows the respective component architecture. This is similar to the simulation architecture, however, our concurrent component scheduler is used and the system executes in real-time.

During development it is recommended to incrementally make small changes and quickly test their eects. The interactive execution mode helps with this


Infantdani Chengwei Wang

Abel Rayan

Greg Eisenhauer

Karsten Schwan

Vanish Talwar

Matthew Wolf

Chad Huneycutt

Infantdani Abel

VScope: Middleware for Troubleshooting Time-Sensitive Data Center Applications

Keywords: Cloud, Data Center, Management, Troubleshooting

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

Introduction

In the 'big data' era, live data analysis applications are becoming easy to scale, as well as being lucrative for or even critical to a company's operation. For instance, by continuously analyzing the live number of page views on its products, an ecommerce website can run a dynamic micro-promotion strategy in which when over 3000 customers are looking at a product for over 10 seconds, an extra 20% discount appears on the web page to increase sales. Other mission-critical examples for e-commerce sites are click fraud and spam detection.

The importance of live data analysis is underscored by the recent creation of real-time or 'streaming' big data infrastructures 1 , which include Flume, S4, Storm, Chukwa, and others [START_REF]Cloudera flume[END_REF][START_REF] Neumeyer | S4: Distributed stream computing platform[END_REF][START_REF] Marz | Twitter's storm[END_REF][START_REF] Rabkin | Chukwa: a system for reliable large-scale log collection[END_REF][START_REF] Condie | Mapreduce online[END_REF][START_REF] Linkedin | [END_REF]15]. Conceptually, these are based Fig. 1. A typical real-time web log analysis application composed from Flume, HBase, HDFS, and Hadoop. In Flume, agents reside in web or application servers, collecting logs and converting them into key-value pairs. Collectors receive and aggregate the local results and insert them into HBase, a distributed, scalable key-value store by which users can query the analysis results on-the-fly. HBase consists of region servers that are equipped with a memory cache, termed MemStore, and a Write Ahead Log (WAL). The data are first written to the WAL and MemStore before being asynchronously persisted to the back-end distributed file system, HDFS, which is typically shared by other data-intensive batch systems, such as Hadoop-based MapReduce codes used for off-line, long-term analyses. Each tier can scale to 1000s of servers or virtual machines.

on the well-established paradigm of stream-or event-based processing [START_REF] Gedik | Spade: the system s declarative stream processing engine[END_REF][START_REF] Abadi | Aurora: a new model and architecture for data stream management[END_REF][START_REF]The evpath library[END_REF], but their attractiveness stems from the fact that they can be easily integrated with other elements of 'big data' infrastructures, such as scalable key-value stores and MapReduce systems, to construct multi-tier platforms spanning thousands of servers or consolidated virtual servers in data centers. A sample platform integrating Flume and other data-intensive systems is depicted in Figure 1.

Crucial to maintaining high availability and performance for these multi-tier applications, particularly in light of their stringent end-to-end timing requirements, is responsive troubleshooting -a process involving the timely detection and diagnosis of performance issues. Such troubleshooting is notoriously difficult, however, for the following reasons:

-Holistic vs. tier-specific troubleshooting. As illustrated in Figure 1, each tier is typically a complex distributed system with its own management component, e.g. HBase or Flume masters. Developed by different vendors and/or managed by different operation teams, tier-specific management can improve the availability of individual tiers, but is not sufficient for maintaining an entire application's end-to-end performance, a simple reason being that issues visible in one tier may actually be caused by problems located in another. Needed are holistic systems to efficiently track problems across tiers.

-Dynamic, per-problem functionality. Problems in complex, large-scale systems arise dynamically, and for each class of problems, there may be different detection, analysis, and resolution methods. Troubleshooting, therefore, is an inherently dynamic activity, involving on-line capabilities to capture differing metrics and to diagnose/analyze them with potentially problemand situation-specific methods [START_REF] Wang | A flexible architecture integrating monitoring and analytics for managing large-scale data centers[END_REF]. -Scalable, responsive problem resolution. In latency-sensitive applications like the one in Figure 1, to maintain desired timing, troubleshooting must be conducted both with low perturbation and with high responsiveness: issues must be detected, diagnosed, and repaired without missing too many events and while maintaining availability for other ongoing actions. -System-level effects. Holistic troubleshooting must extend beyond a single application, to also identify the system-level bottlenecks that can arise in today's consolidated data center or cloud computing systems.

Previous troubleshooting systems have not addressed all of these challenges. Solutions that monitor 'everything all the time' [START_REF] Massie | The ganglia distributed monitoring system: Design, implementation and experience[END_REF][START_REF] Yalagandula | A scalable distributed information management system[END_REF][START_REF] Enterprises | Nagios[END_REF], including both appli- cation and system-level events, do not scale for detailed diagnostics via say, debug-level logging or tracing with consequent high levels of perturbation. This is shown in Figure 2, where continuously logging applicationlevel debugging events on all of its nodes slows down an application's performance by more than 10 times over the baseline. Sampling [START_REF] Ren | Google-wide profiling: A continuous profiling infrastructure for data centers[END_REF][START_REF] Sigelman | Dapper, a large-scale distributed systems tracing infrastructure[END_REF][START_REF] Erlingsson | Fay: extensible distributed tracing from kernels to clusters[END_REF][START_REF] Bhatia | Lightweight, highresolution monitoring for troubleshooting production systems[END_REF] for some of the components and/or for some period of time may not only miss important events, affecting troubleshooting effectiveness, but will also bring about serious performance issues when using a homogeneous and/or random sampling strategy across all nodes, e.g., with Dapper [START_REF] Sigelman | Dapper, a large-scale distributed systems tracing infrastructure[END_REF]'s use of a uniform, low (1/1000) sampling rate. In Figure 2, debug-level logging in the Flume application's HBase tier, the smallest portion of the system (5/122 VMs), results in over 10 times slowdown, which is more than an order of magnitude of the perturbation imposed by debug-level logging in the Flume tier, which has the majority of nodes (95/122). Thus, it is inadvisable to use a high sampling rate for the HBase tier, whereas such a strategy for the Flume tier will likely lead to only modest additional perturbation. An alternative troubleshooting approach chosen by GWP [START_REF] Ren | Google-wide profiling: A continuous profiling infrastructure for data centers[END_REF] is to randomly pick some set of machines. This may work well if that set is in the HDFS tier, but will be prohibitively costly if the HBase tier is picked. Other approaches, like those taken by Fay [START_REF] Erlingsson | Fay: extensible distributed tracing from kernels to clusters[END_REF] and Chopstix [START_REF] Bhatia | Lightweight, highresolution monitoring for troubleshooting production systems[END_REF] to set sampling rates based on the event population, still remain unaware of application level perturbation, resulting in the same issue as the one faced by GWP. We, therefore, conclude that a more flexible system is needed for efficient troubleshooting, where methods can differ for each behavior, tier, and type of analysis being performed.

The VScope middleware presented in this paper makes it possible (1) to adjust and tune troubleshooting dynamically -at runtime -for individual tiers and across tiers, (2) to dynamically deploy any analysis action(s) needed to understand the metric data being captured in the ways required by such troubleshooting, and (3) to do so in ways that meet the perturbation/overhead requirements of target applications. To achieve those ends, VScope, as a flexible monitoring and analysis system, offers the following novel abstractions and mechanisms for troubleshooting latency-sensitive, multi-tier data center applications: [START_REF] Wang | A flexible architecture integrating monitoring and analytics for managing large-scale data centers[END_REF], where we proposed the basic architecture of DPGs and investigated an impact model of metric number/size and various DPG topologies, along with other factors, this paper presents DPG implementation, APIs, and commands, based on which we build VScope's troubleshooting functionality.

VScope's capabilities and performance are evaluated on a testbed with over 1000 virtual machines (VMs). Experimental results show the VScope runtime negligibly perturbs system and application performance, and requires mere seconds to deploy 1000 node DPGs of varying topologies. This results in fast operation for on-line queries able to use a comprehensive set of application to system/platform level metrics and a variety of representative analytics functions. When supporting algorithms with high computation complexity, VScope serves as a 'thin layer' that occupies no more than 5% of their total latency. Further, by using guidance that correlates system-and application-level metrics, VScope can locate problematic VMs that cannot be found via solely application-level monitoring, and in one of the use cases explored in the paper, it operates with levels of perturbation of over 400% less than what is seen for brute-force and most sampling-based approaches.

2 System Design and Implementation

Goals and Non-Goals

The design of VScope is driven by the following goals: (1) flexibility: to initiate, change, and stop monitoring and analysis on any set of nodes at any time, supported by operators for dynamically building and controlling user-defined actions for runtime troubleshooting; (2) guided operation: programmable methods for detecting potential problems and then tracking interactions that may contribute to them, between tiers and across software levels, thereby focusing troubleshooting in ways that can reduce overheads and improve effectiveness; and (3) responsiveness and scalability: to deploy troubleshooting methods with low delay at scales of 1000+ nodes.

VScope is designed to be a general platform rather than a set of ad hoc analysis algorithms/solutions. VScope does not replace operator involvement, but aims to facilitate their troubleshooting efforts. Further, while VScope may be used to seek the root causes of failures, its current implementation lacks functionality like an off-line diagnostic database and a rich infrastructure for determining and using decision trees or similar diagnostic techniques. Also, the methods presently implemented in VScope focus on persistent performance problems that will likely render an application inoperable after some time, i.e., when there are frequent or repeated violations of certain performance indicators that persist if they are not addressed. Having determined potential sources of such problems, VScope can then trigger certain actions for mitigation or recovery, but it assumes such functionality to be supported by other subsystems (e.g., inherent to specific applications/tiers or software levels) or housed in some external system for problem resolution [START_REF] Candea | Microreboot -a technique for cheap recovery[END_REF].

VScope Overview

The system architecture of VScope is depicted in Figure 3(a). The machines (VMs or physical machines) in the target application are managed by a server called VMaster. Operators use VScope operations, DPG commands, or scripts with the DPG API, in a console called VShell provided by VMaster. VMaster executes those commands by deploying DPGs on requested machines to process their monitoring metrics, and it returns results to operators. In detail, it starts a DPGManager to create a new DPG, which essentially, is an overlay network consisting of processing entities named VNodes residing on application machines. The DPGManager dynamically deploys VNodes equipped with assigned functions on specified machines, and connects them with a specified topology. VNodes In VMaster, the metric library defines monitoring metric types and associated collection functions. The function library defines the user-defined and default metric analysis functions, including those used in guidance (see Section 2.5). The above metrics and functions can be dynamically deployed into DPGs for various troubleshooting purposes.

The VScope software stack, described in Figure 3(b), has three layers. The troubleshooting layer exposes basic operations in VShell : Watch, Scope, and Query, which will be described in Section 2.3. The Watch and Scope operations constitute the guidance mechanism, where Watch notifies the operator when and where end-to-end anomalies happen, and Scope provides the potential candidate nodes contributing to the anomalies. Operators (or automated decision engines) can then use Query for in-depth analysis on those candidates yielded by guidance. These operations are built upon the DPG layer. In particular, the guidance mechanism (Watch and Scope) relies on an anomaly detection DPG and on interaction tracking DPGs. The DPG layer also exposes API and management commands to offer finer grain controls and customization. The lowest layer, the VScope runtime, is comprised of a set of daemon processes running on all nodes participating in the VScope system (i.e., the machines hosting the application and additional management machines running VScope). This runtime maintains the connections between machines and implements dynamic DPG creation and management. In virtualized data centers, the VScope runtime can be installed in hypervisors (e.g., Dom0 in Xen), in the virtual machines hosting the application(s) being monitored, in both, and/or in specialized management engines [START_REF] Kumar | vmanage: loosely coupled platform and virtualization management in data centers[END_REF][START_REF] Mansour | I-rmi: performance isolation in information flow applications[END_REF]. Our testbed uses a VScope installation in the Xen hypervisor as well as in the VMs hosting the Flume application.

Troubleshooting Operations

Watch. The Watch operation monitors a list of metrics on a set of nodes2 , and its current implementation applies to them an anomaly detection function in order to detect and report anomalous behaviors for any specified metrics. The parameters of the Watch operation described in Table 1 show its ability to monitor metrics on any VScope node, using detection function specified with detectFunc. Sample functions used in our work include thresholding key performance indicators (KPI), such as request latency and statistics like those based on entropy described in [START_REF] Wang | Online detection of utility cloud anomalies using metric distributions[END_REF]. The frequency and duration of the Watch operation are also configurable. In our Flume application, the Watch operation continuously executes on all the Flume agent nodes, monitoring their end-to-end message latencies and detecting the nodes with latency outliers. Internally, Watch is implemented using an anomaly detection DPG explained in Section 2.5.

Scope. The Scope operation (described in Table 2) discovers a set of nodes interacting with a particular node specified by argument source, at a time specified Argument Description nodeList * a list of nodes to explore graph name of interaction graph source node in interest timestamp * interaction at a specific time distance number of edges direction * backward, forward or both by argument timestamp. This operation guides troubleshooting by informing operators which nodes are related to the problematic node when the anomaly happens. Based on this guidance, operators can deploy a DPG on those nodes (or some subset of them) for further diagnosis, using the Query operation. For instance, for the Flume application, 'horizontal guidance' identifies the HBase region servers with which some specified Flume agent is interacting via a Flume collector, and 'vertical guidance' tracks the mappings between a physical machine and the VMs it hosts. By default, the output of Scope is a list of nodes directly interacting with the source. distance and direction are optional arguments, where the former specifies indirect interactions by setting the value > 1, and the latter specifies the 'direction' of interaction, for instance, 'receiving requests from' or 'sending requests to'.

In a nutshell, Scope works by searching an in-memory, global graph abstraction that describes interactions between every pair of nodes. Multiple types of interaction graphs are supported, covering a range of interactions from event level to network and system levels. These are shown in Table 7 and are specified with the argument graph. The creation and continuous update of the global graph is implemented using an interaction tracking DPG explained in Section 2.5.

Query. The Query function collects and analyzes metrics from a specified list of Though operators can just use VScope operations without knowing the underlying DPG logic, new topologies, new operations and customization of existing functionality can be added easily through direct use of DPG APIs, which is not described in detail here because of space constraints.

Implementation

VScope Runtime. The VScope runtime is implemented with EVPath [START_REF]The evpath library[END_REF], a C library for building active overlay networks. Metric collection uses standard C libraries, system calls, and JMX (at application level). Metrics are encoded in an efficient binary data format [START_REF]The evpath library[END_REF], and a standard format template is used to define new metric types. Built-in metrics and functions are listed in Table 5 and Table 6. As shown in the tables, VScope has a comprehensive set of metrics across application, system and platform levels, and a variety of representative analytics functions that are implemented with standard C libraries and other open source codes [START_REF] De Hoon | Open source clustering software[END_REF]. The DPGs associated with these functions have different topologies. For instance, Pathmap, PCA (Principle Component Analysis) and K-Clustering are implemented as centralized DPGs, as they require global data. of the web log analysis application. This DPG collects the end-to-end latency on each Flume agent, which is defined as the duration between the time when a new log entry is added and the time it (or its associated result) appears in HBase. This is measured by creating a test log entry on each agent, querying the entry in HBase, and computing the difference. The latencies are then aggregated through the tree using Entropy-based Anomaly Testing (EbAT) [START_REF] Wang | Online detection of utility cloud anomalies using metric distributions[END_REF][START_REF] Wang | Ebat: online methods for detecting utility cloud anomalies[END_REF], a lightweight anomaly detection algorithm, to output the agents that are outliers. Other algorithms for anomaly detection and ranking are investigated in [START_REF] Wang | Statistical techniques for online anomaly detection in data centers[END_REF][START_REF] Viswanathan | Ranking anomalies in data centers[END_REF].

Interaction Tracking. Table 7 shows the built-in global graphs supported by

Experimental Evaluation

Experiments are conducted on a testbed running 1200 Xen VMs hosted by 60 physical server blades using Ubuntu Linux (20 VMs per server). Every server has a 1TB SATA disk, 48GB Memory, and 16 CPUs (2.40GHz). Each VM has 2GB memory and at least 10G disk space.

VScope Base Overheads

We install VScope on every VM in a host and vary the number of VNodes on In contrast and as shown in Section 4, heavyweight VScope operations, like those performing tracing or logging may incur considerable overheads, due to the innately high costs of those data collection methods. These facts demonstrate the utility of continuously using the 'thin' VScope layer, which does not add notable costs, and then, only using heavier weight data collection and analysis methods when needed. Further, by having the 'thin' layer point out 'where' and 'when' such heavier weight methods are to be used, the inevitably high overheads of using those methods can be reduced.

DPG Deployment

Fast deployment of DPGs is critical for timely troubleshooting. We evaluate this by measuring the latency for deploying a hierarchical DPG on more than 1000 VMs, each of which has one VNode. The topology has a height of 2, and the total number of leaf VMs varies from 125 to 1000. As expected, Figure 5(a) shows increased deployment times (presented as latency on the Y-Axis) with increased DPG sizes. However, latency remains within 5 seconds even at the scale of 1000 VMs. This would be considered sufficient for current troubleshooting delay requirements stated in [START_REF] Bodik | Fingerprinting the datacenter: automated classification of performance crises[END_REF] (typically 1 hour), but it also suggests the utility of future work on DPG reuse -to use and reconfigure an existing DPG, when possible, rather than creating a new one, or to pre-deploy DPGs where they might be needed. Deploying moderate scale DPGs with hundreds of nodes, however, usually happens within 1 second, suggesting that such optimizations are not needed at smaller scale. Also note that deployment latency varies with different branching factors (bf). At scales less than 750, deploying the DPG with bf 125 has larger latency than those with smaller bf values; this is because parent nodes construct their subtrees in parallel and the parents in the DPG with bf 125 have the biggest subtrees.

Interaction Tracking

The Scope operation relies on efficient methods for interaction tracking. We evaluate a distributed DPG (used for connection graph) by creating a two-level, hierarchical DPG with bf 25. We vary its number of leaves from 125 to 1000, and for this test, each VM has a randomly generated local interaction graph represented by an adjacency list with 1000 vertex entries with each vertex connected to 1000 randomly generated vertices. We measure the total latency from the time the first local graph is generated by leaf VMs to the time when the respective merged graph is created at the root. We also measure the average time of local processing incurred during the per-node aggregation of connection graph information in order to study the dominant factor in total latency.

As shown in Figure 5(b), the total latency for generating a global graph increases as the system scales, but it remains within 4 seconds for 1000 VMs, where each VM has a 1000×1000 local connection graph. This means that the system can generate such a global graph at a resolution of every 4 seconds. Total latency is mainly due to the queuing and dequeuing time on VNodes plus network communication time. This is shown by the small measured local aggregation latency in Figure 5(c). At the same time, since these latencies increase linearly with the total number of inputs, parallel aggregation is a useful attribute to maintain for large scale systems. We also note that the local graphs occupy a fair amount of memory, which suggests opportunities for additional optimizations through use of more efficient internal data structures. Finally, the analytics actions taken by Scope utilize the Pathmap for centralized interaction tracking. In Section 3.4, Figure 6 shows that it can generate a 1000 VM graph within 8 seconds.

In summary, the Scope operation's current implementation is efficient for the long running enterprise codes targeted in our work, but it may not meet the requirements of real-time codes such as those performing on-line sensing and actuation in highly interactive settings like immersive games.

Supporting Diverse Analytics

We use the algorithms in Table 6 as micro-benchmarks to measure the base performance of VScope operations. Tests randomly generate a 1000×1000 matrix of float numbers on each VM, and vary the size of the hierarchical DPG (bf=25) from 125 to 1000 leaf VMs. We measure the latency for analyzing the data on all leaf VMs at each scale. For centralized algorithms, the parent VNodes only relay the data. For the Top-K algorithm, we calculate the top 10 numbers. We conduct K-Means clustering with 5 passes. Figure 6 shows latency breakdowns as well as the total latency of each function. In general, most of the algorithms operate within seconds, with increasing latencies for rising scales. Algorithms with high computational complexity are more costly, of course, but for such 'heavyweight' algorithms, especially for PCA, although the total latencies are over 1.5 minutes at the scale of 1000 VMs, the base VScope implementation contributes only about 4.5% to these delays, and this contribution decreases as the system scales.

Experiences with Using VScope

This section illustrates the utility of VScope for troubleshooting, using the application described in Figure 1 (VScope's DPG architecture was also investigated in other use cases in [START_REF] Wang | A flexible architecture integrating monitoring and analytics for managing large-scale data centers[END_REF][START_REF] Hu | Net-cohort: Detecting and managing vm ensembles in virtualized data centers[END_REF].) The application's Flume tier has 10 collectors, each of which is linked with 20 agents. The HBase tier has 20 region servers, and the HDFS tier has 40 datanodes3 . Experiments use web request traces from the World Cup website [START_REF]Worldcup98 logs[END_REF] to build a log generator that replays the Apache access logs on each of 200 agent VMs. Each agent reads the new entries of the log and sends them to its collector. The collector combines the ClientID and ObjectID as the keyword and the log content as the value, then places the record into HBase. The log generator generates 200 entries per second. The worst case end-to-end latency in the problem-free scenario is within 300 milliseconds.

The VScope runtime is installed on all of the VMs and in addition, on all physical machines (i.e., Xen's Dom0s). In accordance with standard practice for management infrastructures [START_REF] Kumar | vmanage: loosely coupled platform and virtualization management in data centers[END_REF][START_REF] Wang | A flexible architecture integrating monitoring and analytics for managing large-scale data centers[END_REF], one additional dedicated VM serves as VMaster, and 5 dedicated VMs serve as parent VNodes in the two-level hierarchy DPGs used for troubleshooting. Two use cases presented below validate VScope's utility for efficient troubleshooting.

Finding Culprit Region Servers

The first VScope use case crosses multiple tiers of the Flume application. The objective is to find some 'culprit' region server exhibiting prolonged execution times. Those are difficult to detect with standard HBase instrumentation because debug-level logging in region servers to trace their request processing times [START_REF]Hbase log[END_REF] generates voluminous logs and high levels of perturbation to the running server(s). Hence troubleshooting using brute force methods might quickly find a culprit by turning on all of the region servers' debug-level logging and then analyzing these logs (in some central place), but this would severely perturb the running application. Alternative methods that successively sample some random set of servers until a culprit is found would reduce perturbation but would likely experience large delays in finding the culprit server. More generally, for multi-tier web applications, while bottleneck problems like the 'culprit' region server described above commonly occur, they are also hard to detect, for several reasons.

(1) Dynamic connectivity -the connections between the Flume and HBase tiers can change, since the region server to which a collector connects is determined by the keyword region of the collector's current log entry. (2) Data-Driven concurrency -HBase splits the regions on overloaded region servers, causing additional dynamic behavior. (3) Redundancy -a region server is typically connected by multiple collectors. As a result, one 'culprit' region server exhibiting prolonged processing times may affect the end-to-end latencies observed on many agents.

We synthetically induce server slowdown, by starting garbage collection (GC) in the Java Virtual Machine (JVM) on one of the region servers. This prolonged disturbance eventually slows down the Flume agents connected to the region server via their collectors. Experimental evaluations compare VScope, the brute-force, and the sampling-based approaches for finding the culprit region server. The VScope approach follows the 3 steps illustrated in Figure 7. (1) A user at a VShell console issues a Watch operation to find which agents have prolonged endto-end latencies.

(2) Use the connection graph (chosen from Table 7) and the Scope operation to find the connected collectors and the region servers to which they connect. In these guidance actions, the connection graph is the graph parameter, the problematic agent node is the source, and '2' is used as the distance parameter. The output will be the collector and associated region servers. By iteratively 'Scoping' all anomalous agents, we find that they share 5 collectors. Furthermore, the Scope operation returns the set of region servers in use by these collectors, and we can determine that they have 4 region servers in common. Therefore, we select those four as candidate culprits. Under the assumption of only one culprit region server, this operation will succeed because the culprit affects all of these collec-tors. While it will be rare to have multiple culprit region servers in a short period of time, in that case, more candidates may be chosen, but they still constitute only a small set of all region servers.

(3) Here, VScope has narrowed down the search for the problematic region server, and we can now use the Query operation to turn on debug-level logging for the candidates. We note that the region servers yielded by the Scope operation will always include the culprit, because VScope tracks all connections. The user will still have to carefully examine the region server logs to find the problem, but instead of having 20 candidates (the brute-force approach), there are just 4. If the examination is done sequentially (by gathering and examining logs one server at a time) to minimize perturbation, the user can expect to examine 2 logs on the average (requiring 20 minutes of logging and .45GB of data) with VScope, as opposed to 10 logs (requiring 100 minutes of logging and 2GB of data) with the bruteforce approach. If log gathering is performed in parallel to save time, the information provided by VScope allows the retrieval of just 4 logs (0.9GB) vs. 20 logs (4.1GB) by the brute-force approach. Note that, as shown in Figure 8, simultaneously logging on multiple region servers has a non-linear effect upon system performance. Simultaneous logging on only 4 servers (with VScope) slows the overall system down by 99.3%, but logging on all servers (brute-force) slows it by 538.9%. Compromise approaches like random sampling might log on more than one, but fewer than the total number of candidate region servers, hoping to trade off perturbation with 'time-toproblem-discovery'. However, the inherent randomness makes their performance nondeterministic. In contrast, VScope rationally narrows the set of possible bad region servers, thus improving the expected perturbation, log data sizes, and time to resolution in both average and worst cases.

These results validate the importance of VScope's 'guided' operation that explicitly identifies the nodes on which troubleshooting should focus, in contrast to methods that use sampling without application knowledge or that employ non-scalable exhaustive solutions. They also demonstrate VScope's ability to assist with cross-tier troubleshooting. We note that, for sake of simplicity, this use case assumes the root cause to be within the region servers. This assumption can be removed, of course, and in that case, operators can apply further analysis as shown in Figure 7 by iteratively using VScope operations.

Finding a 'Naughty' VM

Previous research has shown the potential for running real-time application in virtualized settings [START_REF] Lee | Supporting soft real-time tasks in the xen hypervisor[END_REF]. However, VMs' resource contention on I/O devices can degrade the end-to-end performance of the application. A typical scenario is that some 'naughty' VM excessively uses a physical NIC shared by other VMs on the same physical host, thereby affecting the performance of the real-time VMs. Potential 'naughty' VMs could be those that run MapReduce reducers and exchange voluminous data with a number of other nodes (e.g. mappers), or those running HDFS datanodes and replicating large files. Contention could also stem from management operations like VM migration and patch maintenance [START_REF] Soundararajan | The impact of management operations on the virtualized datacenter[END_REF].

There are remedies for contention issues like those above. They include migrating the 'naughty' VMs and/or changing network scheduling. VM migration can involve long delays, and changes to VMs' network scheduling may involve kernel reboots that are unsuitable for responsive management. The solution with which we experiment performs traffic shaping for the 'naughty' VM on-the-fly, in the hypervisor, without involving guest VMs. To do so, however, support is needed to first locate the troublesome VM. VScope running in the Dom0s of our virtualized infrastructure provides such support. Specifically, VScope deploys VNodes in each host's Dom0, using the virtualization graph in Table 7 to track mappings between VMs and hypervisors.

We emulate the 'naughty' VM issue by deploying a VM with a Hadoop datanode and tasktracker, on the host where a 'good' VM is running one of the 200 Flume agents. This scenario is chosen to emulate co-running a real-time web log analysis application with a batch system using Hadoop for long term analysis on the data generated by the real-time application. In this case, a problem is created by starting a HDFS benchmarking job called 'TestDFSIO write', which generates 120 2GB files with 4 replicas for each file in HDFS. This 'naughty VM' generates 3 files (we have 40 slaves in the Hadoop configuration. Every slave carries out 3 map tasks, each of which writes a 2G file to HDFS, and replicates them via the network. VScope is used to find that naughty VM, so that its communications can be regularized via Dom0 traffic shaping.

The monitoring traces in Figure 9 demonstrate VScope's troubleshooting process. Trace 1 presents the latency data generated by the Watch operation. Latency rises after the anomaly is injected. Using 1 second as the threshold for an end-to-end performance violation, after 20 violations are observed within 5 minutes, the Watch operation reports an anomaly and its location, i.e., the 'good' VM. After the anomaly is reported, troubleshooting starts for the VM by querying basic VM level metrics, including the number of packages per second represented by Trace 2 4 , where we find that metrics in the VM do not show abnormal behavior. In response, we use the Scope operation to find which physical machine is hosting the VM and then Query its aggregate packet rate. With these guided actions, Trace 3 shows that the shared NIC is exchanging a large number of packets, in contradiction to the low packet rate in the 'good' VM. The next step is to further Scope the virtualization graph to find the other VMs running on the same physical host and then Query the network metrics of their VIFs5 . The 'naughty' VM is easily found, because its respective VIF consumes the majority of the packets for the physical NIC, as shown in Figure 9:Trace 4. The correctness of the diagnosis obtained via VScope is demonstrated by applying traffic shaping in Dom0, which involves using TC to throttle the bandwidth of the 'naughty' VM. It is apparent that this action causes the end-to-end latency of the good VM to return to normal (see Trace 1). In Trace 3, the hypervisor packet rate goes down, and in Trace 4 the network consumption of the 'naughty' VM also sinks, as expected, but it still has its share of network bandwidth.

Related Work

Aggregation systems like SDIMS [START_REF] Yalagandula | A scalable distributed information management system[END_REF] and Moara [START_REF] Ko | Moara: flexible and scalable group-based querying system[END_REF] are most related to VScope in terms of flexibility. SDIMS provides a flexible API to control the propagation of reads and writes to accommodate different applications and their data attributes. Moara queries sub-groups of machines rather than the entire system. In both systems, flexibility is based on dynamic aggregation trees using DHTs (Distributed Hash Tables). VScope's approach differs in several ways. First, VScope can control which nodes and what metrics to analyze; neither SDIMs nor Moara provides this level of granularity. SDIMS only controls the level of propagation along the tree, and Moara chooses groups based on attributes in the query (e.g., CPU utilization). Second, the analysis functions in SDIMS and Moara are limited to aggregation functions, while arbitrary functions can be used with VScope, including those performing 'in transit' analysis. Third, like other monitoring or aggregation systems, including Ganglia [START_REF] Massie | The ganglia distributed monitoring system: Design, implementation and experience[END_REF], Astrolabe [START_REF] Van Renesse | Astrolabe: A robust and scalable technology for distributed system monitoring, management, and data mining[END_REF], and Nagios [START_REF] Enterprises | Nagios[END_REF], SDIMS and Moara focus on monitoring the summary of system state, while VScope's can also be used for in-depth troubleshooting, including debugging and tracing, supported by basic metric aggregation like that performed in the Watch operation.

GWP [START_REF] Ren | Google-wide profiling: A continuous profiling infrastructure for data centers[END_REF], Dapper [START_REF] Sigelman | Dapper, a large-scale distributed systems tracing infrastructure[END_REF], Fay [START_REF] Erlingsson | Fay: extensible distributed tracing from kernels to clusters[END_REF], Chopstix [START_REF] Bhatia | Lightweight, highresolution monitoring for troubleshooting production systems[END_REF] are distributed tracing systems for large scale data centers. VScope is similar in that it can monitor and analyze in-depth system or application behaviors, but it differs as follows. First, instead of using statistical (Fay and Chopstix leverage sketch, a probabilistic data structure for metric collection) or random/aggressive sampling (as used in GWP and Dapper), VScope can look at any set of nodes, making it possible to implement a wide range of tracing strategies (including sampling) through its guidance mechanism. Second, those tracing systems use off-line analysis, while VScope can analyze data on-line and in memory, to meet the latency restriction for troubleshooting real-time applications.

HiTune [START_REF] Dai | Hitune: dataflow-based performance analysis for big data cloud[END_REF] and G 2 [START_REF] Guo | g 2 : a graph processing system for diagnosing distributed systems[END_REF] share similarity with VScope in that they are general systems for troubleshooting 'big-data' applications. HiTune extracts the dataflows of applications, using Chukwa for data collection and Hadoop for dataflow analysis. G 2 is a graph processing system that uses code instrumentation to extract runtime information as a graph and a distributed batch processing engine for processing the queries on the graph. VScope differs in its focus on on-line troubleshooting, whereas HiTune and G 2 are mainly for off-line problem diagnosis and profiling. Further, HiTune and G 2 are concerned with analyzing within a single application tier, while VScope troubleshoots across multiple application tiers. Other troubleshooting algorithms and systems, such as Pinpoint [START_REF] Chen | Pinpoint: Problem determination in large, dynamic internet services[END_REF], Project5 [START_REF] Aguilera | Performance debugging for distributed systems of black boxes[END_REF], and E2EProf [START_REF] Agarwala | E2eprof: Automated end-to-end performance management for enterprise systems[END_REF], target traditional web applications while VScope focuses on real-time data-intensive applications.

Conclusions

VScope is a flexible, agile monitoring and analysis system for troubleshooting real-time multi-tier applications. Its dynamically created DPG processing overlays combine the capture of monitoring metrics with their on-line processing, (i) for responsive, low overhead problem detection and tracking, and (ii) to guide heavier weight diagnosis entailing detailed querying of potential problem sources. With 'guidance' reducing the costs of diagnosis, VScope can operate efficiently at the scales of typical data center applications and at the speeds commensurate with those applications' timescales of problem development. The paper provides evidence of this fact with a real-time, multi-tier web log analysis application.

Our ongoing work is further developing VScope's notion of guided operation, one idea being to automatically generate certain sequences of guidance actions from the previous manual actions taken by operators. We will also investigate other guidance options. To extend scalability to the 10,000+ machines of today's large scale data center applications run by web companies like Google or Amazon, it may also be useful to pre-position DPGs into potentially critical subsystems and/or reconfigure existing DPGs, instead of deploying new ones when investigating problems via detailed queries.

Fig. 2 .

 2 Fig. 2. E2E performance slowdown (i.e. latency increase) % caused by debug-level logging at different tiers of the architecture shown in Figure 1.

Fig. 3 .

 3 Fig. 3. VScope System Design

Fig. 5 .

 5 Fig. 5. Efficient DPG Deployment and Guidance Mechanism

Fig. 6 .

 6 Fig. 6. Analytics Microbenchmark Performance

Fig. 7 .

 7 Fig. 7. Steps using VScope operations

Fig. 9 .

 9 Fig. 9. Using VScope to Find a 'Naughty' VM

1 .

 1 Dynamic Watch, Scope, and Query. VScope abstracts troubleshooting as

	are
	mapped to underlying machines.
	3. Distributed Processing Graphs (DPGs). All VScope operations are realized
	by DPGs, which are overlay networks capable of being dynamically deployed
	and reconfigured on any set of machines or processes, supporting various
	types of topologies and analysis functionalities. First introduced in our pre-
	vious work

a process involving repeated Watch, Scope, and Query operations. Respectively, these (i) detect performance anomalies, (ii) 'zoom-in' to candidate problematic groups of components or nodes, and (iii) answer detailed questions about those components or nodes using dynamically deployed monitoring or analysis functions. VScope can operate on any set of nodes or software components and thus, can be applied within a tier, across multiple tiers, and across different software levels. 2. Guidance. Replacing the current manual 'problem ticket' mechanisms used in industry, VScope based troubleshooting is directed by on-line 'guidance', realized by the Watch and Scope operations that first detect abnormal behavior, followed by exploring candidate sources for such behavior, and only then initiate more detailed queries on select entities. The current implementations of Watch and Scope support both 'horizontal guidance', to track potential problems across different tiers of a multi-tier application, and 'vertical guidance', to understand whether problems are caused by how applications

Table 1 .

 1 Arguments of Watch(*Optional)

	Argument	Description
	nodeList * a list of nodes to monitor
	metricList	a list of metric types
	detectFunc * detection function or code
	duration *	duration
	frequency *	frequency

Table 2 .

 2 Arguments of Scope(*Optional)

Table 3 .

 3 Arguments of Query(*Optional) Query (including query with 'continuous' mode) is designed with the 'pull' model, i.e., the VMaster requests (pulls) metrics/results from VNodes. Conversely, watch is designed with the 'push' model, i.e., VNodes periodically report basic metrics or anomaly detection results to the VMaster.

	Argument	Description
	nodeList * a list of nodes to query
	metricList * a list of metric types
	queryFunc analytics function or code
	mode *	continuous or one-shot

2.4 Flexible DPGs

DPG as the Building Block. All VScope operations described in Section 2.3 are implemented via DPGs. A DPG consists of a set of processing points (VNodes) to collect and analyze monitoring data. It can be configured in multiple topologies to meet varying scale and analysis requirements. For example, it can be configured as a hierarchical tree or as a peer-to-peer overlay or, when operating at smaller scales, as a centralized structure. Managed by a DPGManager, a DPG can be dynamically started on a specified set of nodes, where each VNode runs locally on a designated node and executes functions specified in VScope operations. These functions are stored as binaries in the function library, and they can be dynamically linked. As a result, DPGs are flexible in terms of topology, functions executed, and metric types. Further, DPG outputs can be (i) presented immediately to the VScope user in VShell, (ii) written into rotating logs, or (iii) stored as off-line records in a database or key-value store. The last two configurations are particularly important when historical data is needed for troubleshooting. The use case in Section 4.2 uses rotating logs to store past metric measurements.

DPG API and Management Commands

.

Figure 4 describes the DPG core

Table 4 .

 4 Pseudo Functions for DPG API DPG create (list, topology, spec) Create a DPG with a specified topology int add (src, dst, DPG) Add a link from VNode src to VNode dst int assign (func, spec, list, DPG) Assign function to a list of VNodes VNode src to VNode dst ID for reference in subsequent operations, and in the assign() method, the parameter func is a registered function ID. When a DPG is running, one can call the assign() to change the functionality on any VNode or use the insert() and delete() methods to change the DPG. The DPG API is exposed as commands in VShell, as well, and there are additional auxiliary management commands like list (listing metric types, functions, or DPGs) and collect (returns the metric collection function).

	int start (DPG)	Start a DPG
	int stop (DPG)	Stop an operating DPG
	int insert (new, src, dst, DPG) Insert a new VNode between existing VNodes
	int delete (src, dst, DPG)	Delete a link from

Table 5 .

 5 Basic

		Metrics
	Level	Basic Metrics
	Appli-E2E Latency, JMX/JVM Metrics
	cation Flume/HBase/HDFS INFO Logs
	Virtual	VCPU, Memory, I/O Metrics
	Machine Network Traffic, Connections
	Dom0 & CPU, I/O and Memory Metrics
	System Paging, Context Switch Metrics

Table 6 .

 6 Built-in Functions DPGAlgorithms End Anomaly Detection. The Watch operation is implemented using a DPG with a hierarchical topology in which the leaves are all of the nodes Algorithm 1: Parallel Graph Aggregation 1. On each leaf node, generate an adjacency list (where each record is [vertex ID, connected vertices]) sorted by vertex IDs, and send it to parent 2. On each parent or root node, merge n sorted adjacency lists as follows: i. Create an array P with size of n storing current vertex ID in each adjacency list. ii. If multiple IDs in P are the same and they are the smallest, merge their records into a new record, else take the record with the smallest vertex ID in P as the new record. (essentially an n-way sorting of n vertex ID arrays) iii. Place the new record into the merged adjacency list. iv. Update P to reflect the next record in each adjacency list. v. Repeat ii to iv until all the records in n adjacency lists are visited.

	Watch Hierarchy	MAX/MIN/AVE, Entropy, Top-K
	Scope Centralized	Pathmap[3]
	Query Centralized	K-Clustering, PCA

Table 7 .

 7 VScope Interaction Graphs For interactions like the causality graph implemented using Pathmap[START_REF] Agarwala | E2eprof: Automated end-to-end performance management for enterprise systems[END_REF], a DPG collects metrics from leaves, compresses them at intermediate nodes, and then constructs the graph at the DPG root. An alternate distributed implementation of graph construction uses parallel analysis in which the leaves analyze metrics to generate a local graph (e.g., in the connection graph, it is the ingress and egress connections on a node), the local graphs are aggregated at parent nodes to create partial graphs which are finally aggregated at the root to produce the global graph. The current prototype uses adjacency lists to represent graphs and employs the parallel algorithm shown in Algorithm 1 to merge adjacency lists.

	Scope, covering a range of in-
	teractions from event level to	Interaction	DPG
	network and system levels. For each graph type, in our imple-mentation, a DPG is deployed and continuously run on all the nodes to construct and update the corresponding graph struc-ture in VMaster. There are two ways to track the global interac-tions, centralized or distributed.	Causality Event Flow Graph between Nodes Using Pathmap Centralized Connection Network Distributed Graph Connection Using Netstat Virtual Dom0-DomU Distributed Graph Mapping Using Libvirt Tier Dependency Distributed Graph between Tiers Static Config.

Table 8 .

 8 VScope Runtime Overheads

	each VM. Each VNode collects
	the metrics shown in Table 5,	DPG# VNode# CPU Usage Memory Usage
	and sends them to a separate	in Host in VM	Increase	Increase
	DPG. As shown in Table 8,	20	1	< 0.01%	0.02%
	CPU and Memory overheads	100	5	< 0.01%	0.02%
	to the VM are negligible even	1000	50	< 0.01%	0.03%
	when there are 50 VNodes (1000
	concurrent DPGs in the host). With continuous anomaly detection and via inter-
	action tracking, VScope imposes only 0.4% overhead on the end-to-end latency
	of application described in Section 4.

In this paper we use the term 'real-time' to refer a latency restriction within seconds or hundreds of milliseconds.

A node is a physical or a VM running the VScope runtime in example application.

Each tier has one master node, and in HBase, 5 region servers serve as the ZooKeeper quorum. For simplicity, we do not 'count' masters when discussing scale.

We only show NIC-related metrics for succinctness.

A VIF is the logical network interface in Dom0 accepting the packets for one VM and in our configuration, each VM has a unique VIF.