Pablo Chacin
email: pchacin@ac.upc.edu

Leando Navarro

Utility Driven Elastic Services

Keywords: Web Services, QoS, Epidemic Algorithm, Overlay, Self-adaptive

To address the requirements of scalability it has become a common practice to deploy large scale services over infrastructures of non-dedicated servers, multiplexing instances of multiple services at a fine grained level. This tendency has recently been popularized thanks to the utilization of virtualization technologies. As these infrastructures become more complex, large, heterogeneous ad distributed, a manual allocation of resources becomes unfeasible and some form of self-management is required. However, traditional closed loop control mechanisms seems unsuitable for this platforms. The main contribution of this paper is the proposal of an Elastic Utility Driven Overlay Network (eUDON) for dynamically scaling the number of instances of a service to ensure a target QoS objective in highly dynamic large-scale infrastructures of non-dedicated servers. This overlay combines an application provided utility function to express the service's QoS, with an epidemic protocol for state information dissemination, and simple local decisions on each instance to adapt to changes in the execution conditions. These elements give the overlay robustness, flexibility, scalability and a low overhead. We show, by means of simulation experiments, that the proposed mechanisms can adapt to a diverse range of situations like flash crowds and massive failures, while maintaining the QoS objectives of the service.

Introduction

Modern large scale service-oriented applications frequently address unexpected situations that demand a rapid adaptation of the allocated resources, like flash crowds -that require a quick allocation of additional resources-or massive hardware failures -that require the re-allocation of failed resources. At the same time, applications are expected to maintain certain QoS objectives in terms of attributes like response time and execution cost [START_REF] Menasce | Qos issues in web services[END_REF].

To address these requirements, it has become a common practice to deploy services over large scale non-dedicated infrastructures -e.g. shared clusters -on which servers are dynamically provisioned/decommissioned to services in response to workload variations. In comparison, in a traditional enterprise infrastructure the scale up process takes a long time and requires manual intervention, and therefore over-provisioning services to handle such situations is the common practice.

Chandra et al. [START_REF] Chandra | Quantifying the benefits of resource multiplexing in on-demand data centers[END_REF] demonstrated that fine-grained multiplexing at short timescales -in the order of seconds to a few minutes -combined with fractional server allocation leads to substantial multiplexing gains over coarse-grained reallocations. To accomplish this fine grained multiplexing, it is necessary to count with mechanisms to allocate/deallocate servers efficiently and then be able to manage those servers in a very dynamic environment with a high turn-over of servers.

As these infrastructures become more complex, large, distributed and heterogeneous, some sort of self-adaptive [START_REF] Salehie | Self-adaptive software: Landscape and research challenges[END_REF] (also known as autonomic [START_REF] Kephart | The vision of autonomic computing[END_REF]) capabilities are needed to stabilize its performance within acceptable limits despite variations in the load and resources, recover from failures, and optimize them according to business objectives.

However, as noted in [START_REF] Nallur | Self-optimizing architecture for ensuring quality attributes in the cloud[END_REF] traditional closed loop self-adaptation approaches are of limited applicability in the scenarios described above, as they made a set of restrictive assumptions: a) The entire state of the application and the resources are known/visible to the management component b) the adaptation ordered by the management component is carried out in full and in a synchronized way, and c) the management component gets full feedback of the results of changes made on the entire system. In contrast, in a large scale, wide-area system getting a global system knowledge is infeasible and coordinating adaptation actions is costly. Additionally, each server may belong to different management domains -different sites in an organization, external providers -with different optimization objectives. Two additional problems arise for the self-management of non-dedicated servers: the complexity of eliciting a model to predict the effect of the adaptation decisions and drive the adaptation process, and the need for an isolation mechanism to prevent servers to interfere with each other's performance.

The main contribution of this paper is the proposal of an Elastic Utility Driven Overlay Network (eUDON) for dynamically scaling the number of service instances used to support a service, ensuring a target QoS objective in highly dynamic large scale shared infrastructures.

Based on UDON [START_REF] Chacin | Utility driven service routing over large scale infrastructures[END_REF], eUDON combines a) an application provided utility function to express the service's QoS in a compact way; b) an epidemic protocol for scalable, resilient and low overhead state information dissemination; c) a model-less adaptive admission policy on each service instance to ensure a QoS objective; and d) mechanisms for the elastic assignment of instances to adapt to fluctuations in the load and recover from failures. All these mechanisms act autonomously on each instance based on local information, making the systems highly scalable and resilient.

We show by means of simulations experiments how eUDON adapts to diverse conditions like peaks in the workload and massive failures, maintaining its QoS and using efficiently the available resources.

The rest of the paper is organized as follows. Section 2 presents the general model for eUDON and describes in detail the two main mechanism used to achieve elasticity. Section 3 describes the simulation based experimental evaluation that explores its behavior under diverse scenarios. Section 4 presents relevant work in the field to put the proposed work into context. Finally, section 5 presents the conclusions and outlines the future work.

eUDON: an Elastic Utility Driven Overlay Network

The model for eUDON which is shown in Fig 1. Requests coming from users are processed through a set of entry-points, which correspond to segments of users with similar QoS requirements, and must be routed to service instances that offer an adequate QoS. Requests are evenly distributed over the multiple entry points for the same user segment using traditional DNS or L4 level load balancing techniques [START_REF] Cardellini | The state of the art in locally distributed web-server systems[END_REF]. It is important to notice that in our work we concentrate in the web application layer and assume that the data access, including consistency requirements, are handle by a separated data layer as proposed in modern highly scalable web architectures [START_REF] Kossmann | An evaluation of alternative architectures for transaction processing in the cloud[END_REF].

Each service has a utility function that maps the attributes and execution state of a service instance (e.g. response time, available resources, trustworthiness, reliability, execution cost) to a scalar value that represents the QoS it provides. Utility functions allows a compact representation of the QoS requirements for services and facilitate the comparison of the QoS that different instances provide, even if they run on very heterogeneous nodes [START_REF] Kephart | Achieving self-management via utility functions[END_REF]. The QoS required by a request is defined as the minimum acceptable utility that a service instance must provide to process it. The QoS offered by an instance may vary over time due to, for example, fluctuations on the load or the available resources1 of the non-dedicated server it runs on.

There is large pool of servers available for diverse services. At any given time, on a subset of those servers there are instances activated to process requests for a service. The active instances are organized in a Service Search Overlay, whose objective is to facilitate finding an instance offering adequate QoS. Among the active instances, a subset capable of processing the current workload is promoted -as described later in section 2.2 -to join the Service Routing Overlay. This overlay is used by the entry points to route requests preserving QoS and achieving load balancing. Instances which are underutilized, are demoted and leave the routing overlay but remain in the search overlay. Eventually, instances can be deactivated from the search overlay.

The number of active instances in the search overlay depends on the expected service demand and the level of replication required to ensure resilience to failures as well as to handle short time increases in the demand. To estimate this number approaches like those proposed in [START_REF] Ruth | Viocluster: Virtualization for dynamic computational domains[END_REF] [9] can be applied. In eUDON, this problem is part of our ongoing research and diverse options are being explored. In the rest of this paper we assume the number of active instances is fixed even when the proposed approach can accommodate variations in this number. The main problem we address is how to maintain to a minimum the number of instances in the routing overlay to keep the number of hops needed to process requests low while still offering an adequate QoS.

It is important to notice that in shared clusters, the active instances which are not promoted for processing requests add little overhead to the cluster. In a cloud scenario, it makes sense to have instances activated for some time even if idle, because of the activation overhead and because computing resources are paid by hours -and therefore a 15 minutes activation cost the same than a full hour activation.

The routing and search overlays use a push style epidemic algorithm to maintain their topologies, find new (activated, promoted) instances, remove unavailable (failed, demoted, deactivated) instances, and spread the current state of instances. Periodically each instance selects a subset of its neighbors (the exchange set) and sends a message with its local view (the neighbor set) and its own current state. When a instance receives this message, merges it with its current neighbor set and selects the subset with the more recently updated entries as the new neighbor set. In this way, each instance keeps a local view with the most up date information among all the information received from neighbors.

Next sections describe in detail the two main elements of the model outlined before that provides the attributes of self-adaptiveness and elasticity to eUDON.

Routing

The objective of the routing mechanism is to deliver each request to a service instance that satisfies its QoS requirements, with high probability and the minimal amount of routing hops. In eUDON Requests are routed using the unicast algorithm shown in Fig. 2a. On the reception of a request, the routing algorithm uses an admission function to determine if the instance can accept the request for processing. If the request is not accepted, then a ranking function is used to rank the neighbors and select the most promising one as the next hop in the routing process.

eUDON uses an adaptive admission function -inspired by the one proposed in the Quorum system [START_REF] Blanquer | Quorum: Flexible quality of service for internet services[END_REF] -summarized in Fig. 2b. The utility of the service instance is periodically monitored and compared with a target QoS objective and the size of the admission window is increased or decreased as needed to correct deviations. The only assumption made by this process is that the utility is non-increasing with respect of the load. That is, that increasing/decreasing the load lowers/rises the utility, given that the rest of the utility related attributes remains equal. One significant advantage of this adaptive admission function is that it does not require any model to estimate the future performance of the service. Moreover, it works even when the resources allocated to a service cannot be reserved and therefore the available capacity fluctuates. Every overlay uses a different ranking function. The routing overlay uses a round-robin ranking which has been shown to offer an acceptable performance in this context -see [START_REF] Chacin | Load balancing on large-scale service infrastructures[END_REF] for details. When a request is routed beyond a predefined number of hops, it is routed using the search overlay, looking for an active (but not promoted) instance capable of serving it. This search uses a greedy ranking based on the (last known) utility of the nodes which was shown in [START_REF] Chacin | Utility driven service routing over large scale infrastructures[END_REF] to be highly effective in a wide range of conditions and scenarios.

Promotion and demotion

The decision to join/leave the routing overlay is taken by each instance autonomously based both on its local information, like the rate of request being processed or the server's utilization, and aggregate non-local (potentially global) information like the total workload of the system or the average service rate of other instances.

eDUON uses a probabilistic adaptation mechanism implemented by two rules based on the service rate (the number of service requests processed by time interval). An instance promotes itself if its service rate is close to the average of the system and demotes itself if it is offering a service rate below the 25 percentile of all instances. These rules where chosen for their simplicity and because they could be easily traceable to the system's status, more than for their optimality. However, they exhibit a very acceptable behavior as shown in the experimental results. The probabilistic nature of the rules leads to a progressive adaptation preventing that many instances take simultaneously the same decision, overreacting to a situation and leading to oscillations in the system.

An estimate of the global service rate can be obtained by an epidemic aggregation process embedded into the overlay maintenance algorithms [START_REF] Kempe | Gossip-based computation of aggregate information[END_REF] [START_REF] Jelasity | Gossip-based aggregation in large dynamic networks[END_REF]. In the simulation described in section 3.1, each instance gets an estimated of these values perturbed by an error factor to simulate the estimation error of the distributed aggregation algorithms.

The probability for promotion/demoting an instance is given by:

P (S ) = 1 1 + e kS , (1)
Where S = ( S -S)/ S is the deviation of the node's service rate S from the target service rate S, and k is a parameter that adjust the sensitivity of probability to the service rate. When calculating the probability for demoting, k > 0 and for promoting k < 0. Fig. 3 shows this probability for various values of k, and and S = 50 for promotions and S = 20 for demotions.

As the promotion and demotion rules behaves independently of each other, we have added an additional condition to prevent an instance to be continuously promoted/demoted without having the change to stabilize: instances will not run these rules again for the 3 cycles following a promotion/demotion. This number was empirically obtained by testing multiple options and found to work well in different situations.

Experimental Evaluation

In this section we describe the simulation model and the metrics used for the evaluation, and summarize the results of different experiments developed to analyze how the system adapts to different scenarios.

The results presented correspond, unless the contrary is explicitly indicated, to the average over 10 simulation runs. Each run simulates 200 seconds (300 for the peak load scenario).

Simulation model

We implemented a discrete event simulator for the detailed simulation of the processing of requests allowing us to capture very detailed measurements of the system's behavior. Table 1 summarizes the more relevant simulation parameters.

Overlay. We have simulated an idealized network that mimics a large cluster, on which nodes can communicate with a uniform delay. The base experimental setup was 128 overlay nodes, with a 8 entry points and the 120 service instances (a 1:15 ratio). There is an ongoing work -with promising preliminary results for up to 2048 nodes-to experiment with several thousand instances. However, as the instances work exclusively with local information, we expect that the results will hold for larger scales, as was previously shown in [START_REF] Chacin | Utility driven service routing over large scale infrastructures[END_REF].

All Service instances are initially members of the search overlay, but only a fraction initially join the routing overlay according to a join probability parameter. The adaptation process dynamically adjust this fraction accordingly to the conditions (e.g. load).

Each instance maintains a neighbor set of 32 nodes and contacts 2 of them periodically to exchange information. These values correspond, to the optimal trade off between information freshness and communication costs as discussed in [START_REF] Chacin | Utility driven service routing over large scale infrastructures[END_REF]. Service Instances. Each service instance dispatches requests using a processor sharing discipline. This model fits well for web servers like Apache, a well-known and widely used multi-threaded web server, and is amenable to analytical evaluation using a M/G/1/k * P S queuing system [START_REF] Cao | Web server performance modeling using an m/g/1/k*ps queue[END_REF]. This facilitates the comparison of simulation results with analytical estimates. For instance, the maximum arrival rate that can be processed by the system maintaining a target response time -as below -was estimated using this model.

Arrivals. The service requests arrive following a Poisson distribution and are evenly distributed among the entry points. The arrival rate is calculated using the analytical model for service instances considering the average background load of servers to ensure that the allocation of the workload is feasible, but demands all the available capacity. Therefore, the maximum theoretical allocated demand is of 1.0 and the expected utilization is around 0.95. All requests generated in the tests have the same expected QoS.

Utility Function.

In our experiments we use a utility function that relates the utility to the deviation of the response time RT from a target maximum response time RT 0 :

U (RT ) = RT 0 -RT RT 0 α (2)
As shown in Fig. 4 the coefficient α controls how quickly the utility decreases as the response time approaches the maximum. This function was selected because it can easily be related to metrics obtained from both the analytical and simulation models, making it straightforward to predict and measure the impact of the adaptation decisions in the resulting utility. The evaluation of more complex utility functions is subject of future work. However, it is important to stress that the adaptation process considers only the utility function and makes no explicit reference to the underlying response time. This allows us to generalize the results to other utility functions given that they satisfy the basic assumption of being non-increasing with the load of the system, as discussed in section 2.1. Background Workload One important aspect in our experiments is the evaluation of the impact of background load in non-dedicated servers, which impacts the utility that an instance can provide. This load (defined as a faction of the node's computing power) is modeled as a random variable whose value varies along the time following a random walk with certain Variability. This model is consistent with the observed behavior of the load on shared servers [START_REF] Oppenheimer | Service placement in a shared widearea platform[END_REF][START_REF] Yang | Homeostatic and tendency-based cpu load predictions[END_REF].

Metrics

In the evaluation of the proposed mechanisms we have considered the following metrics, which are related to its main objectives of efficiently delivering requests to an appropriate node to maintain an adequate QoS: Allocated Demand: measures the fraction of the demand that is actually allocated to a server, before being dropped due to the expiration of its TTL (set to 8 hops in our experiments). This metric measures how effective is the systems in allocating requests.

QoS Ratio: is the ratio between the target QoS expected by a service request and the actual QoS received. A ratio below 1.0 means target was not met, while a ratio over 1.0 means the target was exceeded (which is not necessarily desirable, as it may indicate the server is underutilized).

Utilization: Measures the percentage of the node capacity being used, considering both the background load and the load produced by the service requests. This metric is relevant as measures how efficiently resources are used.

Routing hops: measures the number of hops (or retries) needed to allocate a request to a server with an adequate utility. It measures how efficient is the mechanism in allocating requests.

In the graphics of experimental results in section 3.3 the percentiles 25, 50 and 75 of the QoS Ratio, Utilization and Routing Hops are presented to show the variability of these metrics. Percentiles 25 and 75 are drawn as a filled curve and percentile 50 as a continuous line.

Results

In this section we describe the different experiments we made to test the behavior of the system under diverse conditions and usage scenarios.

Base scenario. In the base scenario, a steady workload is submitted to the system that demands all the available capacity to achieve the QoS objective. Fig. 5a shows how the utility driven adaptation occurs for an instance. As the background load fluctuates so does the capacity of the instance -following the adaptive admission described in section 2.1 -to compensate the change in the available CPU and maintain the QoS ratio close to 1.0. With respect of the overall behavior of the system, as shown in Fig. 5b it quickly converges to a utility ratio of 1.0, with a small variability. The adaptation process also achieves a high level of system efficiency, with a total utilization of system capacity around 0.9. Additionally, 90% of the maximum theoretical workload is allocated -this figure is maintained in all scenarios -and a 75% of request needs at most 3 hops to be allocated (graphics for these results not shown for brevity). These results show that the system is effective in achieving the QoS goals, efficient in the utilization of resources and imposes a low overhead in terms of the number of hops needed for allocation.

Peak load scenario. In this scenario, the system is initially submitted to a steady workload that demands 70% of the the available capacity, but at time 100s, an additional load is injected. As can be seen in Fig. 6, the systems quickly reacts to the load by promoting more instances. The overall utilization of the system also is increased -the percentile 25 of the Utilization rises significantly -but the QoS Ratio is maintained during this adaptation process. At time 200s, the additional load is removed and the systems returns to the previous state, demoting the instances not longer needed. Failure scenario. In this scenario, the system is submitted to a steady workload that demands a fraction of its capacity. At time 100s, 20% of the promoted instances fail -a correlated failure as expected in clusters. Fig. 7 shows how the system reacts, incorporating more instances until the system stabilizes. As can be seen, the utility ratio is maintained along this process -except for a short period just after the failure -as requests are routed to nodes in the search overlay; as a consequence, routing hops increase until all the required nodes are promoted to the routing overlay.

Related Work

The elasticity in the allocation of resources to web applications has attracted significant attention from different perspectives.

In [START_REF] Karve | Dynamic placement for clustered web applications[END_REF] the dynamic placing of the instances of multiple applications on a set of server machines is formulated as a two dimensional packaging problem and several heuristics are proposed to optimize the solution by minimizing the number of placement changes while maximizing the balancing of the load across nodes. However solving this problem has a high computational complexity, severely limiting its scalability.

VioCluster [START_REF] Ruth | Viocluster: Virtualization for dynamic computational domains[END_REF] uses both machine and network virtualization techniques which allow a domain in a shared cluster to dynamically grow/shrink based on resource demand by negotiating the borrowing/lending of virtual machines with other domains. This approach has, however, a significant overhead for both the negotiation process and the need to create and start machines.

In [START_REF] Shen | Integrated resource management for clusterbased internet[END_REF] a utility related performance metric is used by a request scheduler to order the processing of requests from multiple services classes, so that the resulting aggregate utility is maximized. The main drawback of the proposed schema is its dependency of a cluster level centralized load balancing, making it unpractical to the scales of our systems of interest. Also, it requires the on-line elicitation of the resource consumption profile for each service -using application supplied metrics-to adjust the resource allocation, while our approach uses a model-less adaptation.

Closer to out work, in [START_REF] Babaoglu | Managing clouds: a case for a fresh look at large unreliable dynamic networks[END_REF] nodes are self-organized and sliced according to an application defined metric and the group that represents the "top" slice are selected to form the application's overlay. Its main drawback is that as the nodes' attributes may change continuously the slicing must also be continuously updated, and operation that require the execution of protocols that run over "epochs" of several update cycles in the order of several seconds. This re-quirements make this approach unsuitable to the scenarios of interest. In our approach, we integrate this process of updating the set of active nodes into the routing process, making it more responsive to changes. Besides, there's no empirical evidence of the actual performance of the proposed model.

Conclusions

We have presented eUDON an overlay for dynamically scaling services on large scale infrastructures of non-dedicated servers. The evaluation of the different scenarios shows the ability of eUDON to adapt to changing conditions using only local information and local decisions, while maintaining the QoS objectives and a high utilization level. More importantly, the system is highly scalable and resilient to failures, two characteristics that are critical for systems based on commodity hardware clusters.

A salient feature of eUDON is its model-less adaptation approach, which can be used in scenarios where there is not a model to predict the QoS of a service, or applying such a model is not feasible due to the dynamism of the environment. Moreover, the adaptation does not require any isolation between competing services.

One additional advantage of the proposed model is that it unifies different events, like service failure, saturation or demotion under a single set of simple adaptation mechanism, simplifying the system design.

The results presented are part of a work in progress and there are still diverse aspects to develop. As already mentioned, the continuous adaptation of the number of active instances (the activation/deactivation mechanism) is still an open issue which is being actively researched. We envision using a mechanism similar to the one used for promotion/demotion but triggered at the servers to decide which services to activate/deactivate.

Additionally, we use only instantaneous measurements of an instance's utility for the various adaptation decisions and in particular, for the admission acceptance window. As a result, even when the average utility ration is around 1.0 and shows little variability, no guarantees of the type "95% of request will have a certain QoS" are currently possible. We are exploring the utilization of a form of summarization of the recent history to offers such guarantees.

Fig. 1 :

 1 Fig. 1: Elastic service overlay model.

  (a) Routing process (b) Adaptive admission.

Fig. 2 :

 2 Fig. 2: Adaptive routing.

Fig. 3 :

 3 Fig. 3: Promotion/Demotion probability function for diverse values of k.

Fig. 4 :

 4 Fig. 4: Utility Function.

  Utilization and QoS Ratio.

Fig. 5 :

 5 Fig. 5: Behavior for base scenario.

  Utilization and Utility Ratio.

Fig. 6 :

 6 Fig. 6: Behavior for peak load scenario.

  Utilization and Utility Ratio.

  Number of instances and Number of Hops.

Fig. 7 :

 7 Fig. 7: Behavior for failure scenario.

Table 1 :

 1 Simulation parameters.

	Parameter	Values	Description
	Servers	128, ... 2048	Number of instances
	Entries ratio	1:15	Ratio of entry points, with respect of
			the number of instances
	Neighbor set	32	Number of neighbors maintained per
			node in the overlay
	Update cycle	1	Frequency of information dissemina-
			tion (in seconds)
	Exchange set	2	Number of neighbors contacted per
			node on each update cycle
	Adaptation cycle 3	Frequency of the adaptation process (in
			seconds)
	Join probability .60	Probability of an service instances to
			join the routing overlay at initialization
	Load Maximum .5	Maximum fraction of a server capacity
			used by background load
	Load variability 0.10	Maximum variation of background load
			per second
	QoS	0.7	Target utility for requests
	K	-3 (promotion)	Adaptation probability adjust constant
		3 (demotion)
	S	50 (promotion)	Target service rate for promo-
			tion/demotion
		20 (demotion)

This work has been partly supported by Spanish MEC grant TIN2010-20140-C03-01.

For example, modern virtualization technologies make feasible to change resource assignments to execution environments on the fly.


Lazaros Koromilas
email: koromil@ics.forth.gr

Kostas Magoutis
email: magoutis@ics.forth.gr

CassMail: A Scalable, Highly-Available, and Rapidly-Prototyped E-Mail Service

In this paper we present the design and implementation of a scalable e-mail service over the Cassandra eventually-consistent storage system. Our system provides a working implementation of the SMTP and POP3 protocols and our evaluation shows that the system exhibits scalable performance, high availability, and is easily manageable under write-intensive e-mail workloads. The design and implementation of our system is centered around a synthesis of interoperable components for rapid prototyping and deployment. Besides offering a proof of concept of such an approach to prototyping distributed applications, we further make two key contributions in this paper: First, we provide a detailed evaluation of the configuration and tuning of the underlying storage engine necessary to achieve scalable application performance. Second, we show that the availability of scalable storage systems such as Cassandra simplifies the design and implementation of higher-level scalable services, especially when compared to the effort expended in projects with similar goals in the past (e.g., Porcupine). We believe that the existence of infrastructural services such as Cassandra brings us closer to the vision of a universal toolbox for rapidly prototyping arbitrary scalable services.

Introduction

The large-scale adoption of the Internet as a means of personal communication, societal interaction, and a successful venue for conducting business has raised the need for applications and services that can support Internet-scale communities of users. Cloud computing [START_REF] Armbrust | Above the Clouds: A Berkeley View of Cloud Computing[END_REF] has recently offered the infrastructural support necessary for small, medium and large enterprises alike to deploy such services. However both application developers and Cloud computing providers are in need of the infrastructural services and platforms that can support the scalability requirements of distributed applications. Compounded with the need for scalability is the need for rapid prototyping. Today's planet-scale social networks such as Facebook, and application marketplaces such as AppleStore have brought applications (and the "garage innovator" behind them [START_REF] Elson | Handling Flash Crowds from your Garage[END_REF]) closer to large communities of users. This trend has fueled a race among developers to bring new ideas to market as soon as possible without sacrificing scalability and availability along the way. The combination of the above needs (scalability and rapid prototyping of application-specific designs) was certainly a challenging undertaking a decade ago. As the core theme of this paper indicates however, recent advances in scalable infrastructure services have improved on the state of the art.

In this paper we substantiate the above observations by focusing on a specific case study: scalable e-mail services. E-mail is an important application offered as an Internet-accessible service by companies such as Google (Gmail), Yahoo (Yahoo! Mail), and Microsoft (Hotmail) among others. Constructing a scalable, highly-available e-mail service has in the past been performed in a variety of ways, by either statically partitioning users and their data in specific machines [START_REF] Christenson | A Highly Scalable Electronic Mail Service using Open Systems[END_REF] using a general-purpose distributed file system as an underlying scalable store [START_REF] Ghemawat | The Google File System[END_REF][START_REF] Thekkath | Frangipani: a Scalable Distributed File System[END_REF] or by specifically designing and constructing an entire system from first principles for the targeted application [START_REF] Saito | Manageability, Availability, and Performance in Porcupine: A Highly Scalable, Cluster-based Mail Service[END_REF]. Whereas the first approach results in simpler systems, experience shows that it suffers from scalability issues specifically in the areas of load balancing (due to static partitioning) and availability (due to strong consistency built into general purpose file systems). The latter approach has in the past been shown to address the above issues, however at the cost of significant system engineering to support (i) fine-grain, balanced data partitioning and (ii) a weaker consistency model that matches the semantics of e-mail protocols. The system presented in this paper combines for the first time the best of both approaches: A synthesis of interoperable components resulting in a simpler system, capitalizing on standardized support for (i) and (ii) above.

A key motivation for this paper is the observation that in recent years there has been significant interest in developing, and in many cases open-sourcing, scalable infrastructural services. This activity has culminated into systems that offer storage/file and data APIs with strong [START_REF] Chang | Bigtable: A Distributed Storage System for Structured Data[END_REF][START_REF] Ghemawat | The Google File System[END_REF][START_REF] Shvachko | The Hadoop Distributed File System[END_REF] as well as weak [START_REF] Hastorun | Dynamo: Amazon's Highly Available Key-Value Store[END_REF][START_REF] Lakshman | Cassandra: a Decentralized Structured Storage System[END_REF][START_REF] Vogels | Eventually Consistent[END_REF] consistency semantics. A prime example of the latter class of storage systems that is used in this paper is Apache Cassandra [START_REF] Lakshman | Cassandra: a Decentralized Structured Storage System[END_REF], a scalable, highly-available, eventually-consistent key-value store originally developed by Facebook [START_REF] Lakshman | Cassandra: a Decentralized Structured Storage System[END_REF]. The existence of Cassandra prompted us to revisit the question of how would one design and build a scalable e-mail service over an eventually-consistent replicated storage system today. More specifically, we considered the following questions:

1. Does Cassandra significantly reduce the development effort compared to the effort expended in a project with similar goals in the past [START_REF] Saito | Manageability, Availability, and Performance in Porcupine: A Highly Scalable, Cluster-based Mail Service[END_REF]?

2. Does the resulting system exhibit the scalability and availability expected of a robust scalable e-mail service?

More generally, our work puts novel infrastructural services such as Cassandra into context with past efforts to explore the feasibility and utility of providing high-level abstractions or data structures as the fundamental storage infrastructure. The Boxwood project [START_REF] Mccormick | Boxwood: Abstractions as the Foundation for Storage Infrastructure[END_REF] and the scalable distributed data structure (SDDS) approach of Gribble et al. [START_REF] Gribble | Scalable, Distributed Data Structures for Internet Service Construction[END_REF] are two examples of such efforts. We believe that a key missing piece in past proposals are primitives that explore the space of data consistency semantics. This paper shows that Cassandra is such a missing piece that brings us closer to the vision of a toolbox of universal abstractions to support arbitrary scalable application development.

Our contributions in this paper are:

-The design and implementation of a fully functioning scalable, highly-available e-mail service based on synthesis of interoperable components (extensible high-level development interfaces, Cassandra storage system).

-Demonstration of the speed of prototyping that such a software engineering approach allows. Specifically, it took us a few tens of lines of Python code to implement a working prototype compared to the 41,000 lines of C++ code it took for a system with similar design principles a decade ago [START_REF] Saito | Manageability, Availability, and Performance in Porcupine: A Highly Scalable, Cluster-based Mail Service[END_REF].

-Evaluation of the configuration and tuning of the underlying storage engine for the targeted application, exhibiting the scalabilibity, availability, and manageability properties of our rapidly-prototyped system.

The rest of the paper is organised as follows. We refer to related work in Section 2. In Section 3 we describe the system architecture and in Section 4 we provide the details of our implementation. In Section 5 we describe the evaluation of our system. In Section 6 we discuss possible optimizations and finally, in Section 7 we conclude.

Related work

The penetration of e-mail in our way of daily life is such that it is nowadays a mandatory offering by all Internet-scale service providers to their subscribers [START_REF] Ducheneaut | E-mail as habitat: an exploration of embedded personal information management[END_REF]. Large-scale e-mail services have in the past been implemented in a number of ways. Early distributed e-mail services partitioned e-mail state across a number of storage nodes but did so in a static partitioning scheme using distributed file systems [START_REF] Christenson | A Highly Scalable Electronic Mail Service using Open Systems[END_REF]. Such a scheme is hard to manage (in particular, it is hard to rebalance e-mail to storage node in case of failure, or to correct an initially unbalanced partitioning). Cluster-based e-mail services [START_REF] Vogels | Scalability of the Microsoft Cluster Service[END_REF] attempted to achieve scalability and availability via database failover schemes with limited success. Finally, application-specific designs [START_REF] Saito | Manageability, Availability, and Performance in Porcupine: A Highly Scalable, Cluster-based Mail Service[END_REF] achieved better scalability via the use of hash-based partitioning of user e-mail to storage nodes, optimistic replication [START_REF] Vogels | Eventually Consistent[END_REF], and dynamic membership algorithms. Such approaches however were complicated and thus were met with little practical success in term of real-world deployment and use.

CassMail shares the basic premise behind systems such as Porcupine [START_REF] Saito | Manageability, Availability, and Performance in Porcupine: A Highly Scalable, Cluster-based Mail Service[END_REF], namely that the semantics of e-mail protocols are naturally relaxed and users are used to e-mail being occasionally delayed, reordered, reappearing after being deleted, or even lost. Thus it is based on a storage system (Cassandra [START_REF] Lakshman | Cassandra: a Decentralized Structured Storage System[END_REF]) utilizing optimistic replication to achieve scalability and high availability. Cassandra is an eventually-consistent storage system, meaning that replicas can temporarily diverge (and clients allowed to see inconsistent intermediate state) but guaranteed to eventually converge. Based on a general-purpose eventually-consistent key-value store rather than its own implementation, CassMail leverages a robust and tested scalable software component and at the same time radically simplifies the overall architecture focusing on the core logic of the application.

The idea of using foundational storage abstractions to support applicationspecific services is not new. Boxwood [START_REF] Mccormick | Boxwood: Abstractions as the Foundation for Storage Infrastructure[END_REF] explored the idea in which high-level abstractions can facilitate the creation of scalable and fault-tolerant applications. Boxwood proposed a set of components (replicated logical devices, a chunk data store, and a scalable B-tree, along with a set of infrastructure services such as global state management, lock service, and a transactional service) as a comprehensive toolbox for distributed system design. At a smaller scale, scalable distributed data structures [START_REF] Gribble | Scalable, Distributed Data Structures for Internet Service Construction[END_REF] proposed a key-value hash table as another foundational abstraction for the support of scalable applications. A common theme in the above proposals was the assumption of strong consistency semantics (singlecopy serializability [START_REF] Papadimitriou | The serializability of concurrent database updates[END_REF]), which in some cases limit system availability and may be constraining applications, especially when their semantics do not strictly require it. Components such as Cassandra extend and enrich the above proposals.

Design

We will first give a brief overview of the Cassandra data model and the schema that we designed for CassMail. Cassandra's basic data unit is the column, or block as we will refer to it next. A block consists of a key and a value. Sequences of blocks (an arbitrary number) collectively form a row. Blocks in a row can be ordered in a user-specified manner depending on key type (for example in timestamp order). Each row is identified by a separate key. A row is individually placed on a Cassandra storage node based on a consistent hashing scheme [START_REF] Lakshman | Cassandra: a Decentralized Structured Storage System[END_REF] described later in this section. Rows are grouped into column families, which are entities akin to relational database tables. Column families are grouped into keyspaces.

Figure 1 displays the schema in which Cassandra stores user and mailbox information. There are two tables, Mailboxes and Users, within a keyspace called Mail. Users is used to validate a user (the origin or destination of an e-mail message) and to find the names of his mailboxes. Each row in Users is keyed by user name. The blocks stored in a row have as their keys the names of the user's mailboxes. Concatenating a username with a mailbox name forms a natural key to a row in the Mailboxes table. The row contains blocks that hold the actual e-mail messages in that user's mailbox. The key for each block is a time-based universally unique identifier (UUID) stamped by the SMTP daemon when the message arrives and is stored. The value of a block is the e-mail message itself. We chose not to fragment a user's mailbox across multiple rows (as for example Porcupine does [START_REF] Saito | Manageability, Availability, and Performance in Porcupine: A Highly Scalable, Cluster-based Mail Service[END_REF]) for two reasons: First, we believe that spreading the load of retrieving a mailbox to several nodes can be achieved by reading different block ranges from different replicas of the row rather than different fragments of the mailbox. Second, we want to avoid hard-to-adjust magic constants (such as the soft limit used by Porcupine [START_REF] Saito | Manageability, Availability, and Performance in Porcupine: A Highly Scalable, Cluster-based Mail Service[END_REF]) to restrict the mailbox spreading too far across storage nodes. Cassandra runs on a cluster of n storage nodes as shown in Figure 2. Each node maps to a specific position on the ring through a hash function. Similarly, each row maps to a position on the ring by hashing its key using the same hash function. Each node is in charge of storing all rows whose keys hash between this node's position and the position of the previous node on the ring. Cassandra members communicate with each other to exchange membership and liveness information through RPC calls. They also communicate when looking for the node in charge of the client's requested data. In our design each Cassandra node is also running an SMTP and a POP3 server. In this fashion, the cluster consists of functionally identical nodes that can perform any and all functions. This symmetrical configuration underlies the system's scalability and availability properties.

An example of e-mail delivery and retrieval is depicted in Figure 2. In this example anne connects1 through her mail-submission agent (MSA) to the SMTP server on node 1 to send an e-mail message to pete. The SMTP server inspects the message and saves it to pete's inbox and anne's outbox on Cassandra. Now suppose that pete wants to check his messages by accessing node 2. He connects to the POP3 server on node 2 with his mail-retrieval agent (MRA) which first receives the number of e-mails (equal to the number of columns in the pete:mailbox row) then asks for message headers, and finally retrieves a number of e-mail messages. POP3 fetches pete's mailbox from Cassandra in fixed-size batches. Each batch would normally go to a different replica for the row to ensure that the read load is balanced across the Cassandra cluster. Eventually pete receives the new message from anne.

Implementation

In this section we describe our implementations of the SMTP and POP3 servers and the configuration and tuning of the Cassandra system to support our writeintensive e-mail workloads. The set of commands implemented by the SMTP and POP3 servers is shown in Table 1. This is the minimum set that enables Mail User Agents (MUAs) to properly receive and send e-mail. We initially implemented the SMTP and POP3 servers in Ruby using the generic GServer class2 . We found that Gserver handles many low-level management tasks, allowing the developer to focus on the specifics of the SMTP and POP3 protocols. Additionally, the Ruby client for Cassandra3 provides a clean, high-level interface that is easy to work with. However, our preliminary experiments showed that the underlying implementation of Gserver was not robust enough to successfully pass our stress tests. Given this early experience we decided to switch to using Python's smtpd module and the pycassa4 client library for Cassandra which proved to be a more robust and performant implementation. Our working implementation of the SMTP and POP3 servers consists of a few tens of lines of code that is easy to reason about and extend.

In addition to implementing the POP3 protocol, we extended the implementation to support multiple mailboxes per user (a feature not directly supported by POP3). We achieved this by appending a delimiter to the username, followed by the specific mailbox name, as for example in POPUSER="anne|outbox". Upon receiving such a name the POP3 server extracts the username and mailbox (using inbox as the default) and uses it to interact with Cassandra.

SMTP and POP3 servers access Cassandra through the Thrift5 interface. Thrift transparently handles Cassandra node availability issues such as failing over to another Cassandra node when the current one appears to have failed. In our implementation we collocate SMTP, POP3, and Cassandra servers on each system node (thus exposing Thrift on the localhost interface). This was a design choice we took to arrive at a homogeneous system in which any node can perform any task. However it would be a trivial modification to enable SMTP/POP3 servers to access Cassandra over the network thus decoupling them into two separate tiers.

Properly configuring Cassandra is key for tuning the system towards specific workloads and environments. Our schema described in Section 3 can be embodied within a very concise description that comprises the name of the keyspace, the column families included, and the information of how to sort the blocks inside them, as shown in the following configuration excerpt: keyspaces:

-name: Mail replication_factor: 3 column_families:

-name: Mailboxes compare_with: TimeUUIDType -name: Users compare_with: BytesType

Another configuration decision is how to partition the logical ring between Cassandra nodes. Our experience with automatic/random node placement on the ring is that it can lead to hot-spots. We thus opted for precomputing initial tokens for our nodes ensuring that each node gets roughly equal key ranges. We used Cassandra's RandomPartitioner6 to achieve balanced distibution of row keys on the ring by hashing them using the MD-5 hash function.

In case of a node failure, surviving nodes detect the failed node and exclude them from their membership list. Cassandra does not automatically repartition the ring to the surviving nodes. This is something that requires manual intervention with the execution of shell commands by an operator. However the system (depending on settings described below) can be available while operating under failure as our evaluation in Section 5 shows.

Two other key Cassandra parameters are the degree of replication (or replication factor) for row data and the level of consistency chosen for reads and writes. Describing the full set of options offered by Cassandra is outside the scope of this paper (see [START_REF] Hastorun | Dynamo: Amazon's Highly Available Key-Value Store[END_REF][START_REF] Lakshman | Cassandra: a Decentralized Structured Storage System[END_REF] for a complete discussion). We will however describe the options we exercised during our evaluation to highlight the key tradeoffs involved. In terms of consistency levels, we used the following conditions for acknowledging a write: ONE Write must be stored at the memory table and commit log of at least one replica. ALL Write must be stored at the memory tables and commit logs of all replicas.

Our implementation uses Cassandra7 version 0.7.0 running on Java 1.6.0 22. We used Ruby version 1.9.2 and Python version 2.6.6.

Evaluation

In this section we report on our experimental results. Our experimental setup consists of a 10-node cluster of dual-CPU AMD 244 Opteron servers with 2GB DRAM running Linux 2.6.18 and connected through a 1Gbps Ethernet switch using Jumbo (9000-byte) frames. Each node was provisioned with a dedicated logical volume comprising four 80GB SATA disks in a RAID-0 configuration. We used the xfs filesystem on this volume on all nodes. The benchmark used in this study was Postal8 , a widely-used benchmark for SMTP servers. Postal operates by repeatedly and randomly selecting and e-mailing a user (USER@example.com) from a population of users. We created a realistic population of users by using the usernames from our departmental mail server (about 700 users). The Postal client is a multithreaded process that connects to a specific SMTP server. In our experiments we configured Cassandra with different replication factors (1, 2, 3) and consistency-levels (ONE and ALL described in Section 4). We used message sizes drawn uniformly at random from a range of sizes. We experimented with two ranges:

-200KB-2MB (typical of large attachments) -50KB-500KB (typical of small attachments) These ranges are chosen to reflect the increase in average e-mail size compared to a related study performed a decade ago [START_REF] Saito | Manageability, Availability, and Performance in Porcupine: A Highly Scalable, Cluster-based Mail Service[END_REF] due to widespread use of attachments in everyday communications.

Each experiment consists of an e-mail-sending session blasting the CassMail cluster to saturation for about ten minutes. Our measurements are per-minute Postal reports of the sum of e-mail data sent during the previous minute (that is e-mail payload, excluding other control/header information). In order to avoid taking into account any bootstrapping overhead we only consider the last five minutes in our measurements. In all of our graphs we report aggregate average throughput and standard deviation (as error bars) of our measurements. Each node ran an instance of the SMTP server (Python code) and an instance of the Cassandra server (Java code), with each server consuming one of the two CPUs. In all cases performance is limited by the servers' CPUs. There was also swapping and garbage collection activity taking place during runs. We consider such activities unavoidable (especially when running software in high-level, scripted, and garbage collected languages such as Python and Java) and legitimate part of a node's load. We used two dedicated client machines with similar specifications to our servers to drive all experiments. The client machines hosted Postal processes in a setup that balanced load-generation work across the two machines. Our first experiment measures the aggregate write throughput over increasing cluster sizes for messages in the range 200KB-2MB. Our results are depicted in Figure 3. Lighter bars correspond to higher replication factors (1-3). The consistency level is set to ONE in all cases. Increasing cluster size results into higher aggregate throughput across all replication factors. The performance increase is smaller going from 1 to 2 nodes due to the introduction of Cassandra server-toserver traffic to forward keys to the proper coordinator (since the e-mail clients are unaware of the mapping between user mailboxes and Cassandra nodes). Increasing the replication factor results into decreased throughput by about 5-10% for each extra replica at all cluster sizes due to the additional traffic necessary to update replicas. We expect this drop to be steeper for stricter consistency levels such as ALL. Note that a replication factor of 2 and 3 does not make sense for cluster sizes of 1 and 2 respectively, explaining the missing bars in Figure 3. Figure 4 depicts the impact of the consistency level in aggregate write throughput with increasing cluster sizes (2-8) and replication factors [START_REF] Chang | Bigtable: A Distributed Storage System for Structured Data[END_REF][START_REF] Christenson | A Highly Scalable Electronic Mail Service using Open Systems[END_REF]. The key observation is that stronger consistency requirements (ALL instead of ONE) degrade performance in all cases. The degradation is more pronounced at larger cluster sizes and is about 35% in the case of 2 replicas (dropping from 50MB/s to 30MB/s) and about 45-50% in the case of 3 replicas. A key factor responsible for this degradation is the large imbalances in nodes' performance due to background tasks such as Java garbage collection or swapping activity. These imbalances are largely masked at consistency level ONE but exposed to the clients at consistency level ALL. This observation highlights a key advantage of eventually-consistent storage systems compared to strongly-consistent ones under write-intensive workloads. Previous work [START_REF] Gribble | Scalable, Distributed Data Structures for Internet Service Construction[END_REF] has pointed out the adverse impact of garbage collection activity in strongly-consistent storage systems written in Java, namely stalling write operations when one out of a group of replicas freezes while undergoing some background activity. Eventually-consistent systems can hide that stall time by allowing operations to progress at the speed of the fastest replica.

We next focus on the impact of message size on aggregate throughput. Figure 5 depicts system performance with increasing cluster size at consistency level ONE and replication factor 1. We observe a performance drop when mov- ing from larger (200KB-2MB) to smaller (50KB-500KB) e-mail messages, which is about constant in absolute value (≈5MB/s) but decreases in relative terms with increasing cluster size. This drop is caused by the higher impact of peroperation overheads (connection setup/teardown, header information generated and processed for each e-mail message, etc.). We next explore the availability of CassMail service when experiencing node failure under replication factor 2 at different consistency levels (ONE, ALL). Figure 6 depicts aggregate per-minute throughput before (0 -3 ) and after (4 -10 ) failure for consistency level ONE (squares) and ALL (dots). For this experiment we used an 8-node CassMail cluster in which all nodes run Cassandra servers and only four out of them also run SMTP servers. At minute 3 we inject a crash failure on one of the Cassandra-only nodes. In the case of relaxed consistency (ONE, squares) the node failure has no apparent effect on performance since the surviving replica takes the update (and thus all writes completing successfully without delay) while failover mechanisms (such as hinted handoff [START_REF] Hastorun | Dynamo: Amazon's Highly Available Key-Value Store[END_REF][START_REF] Lakshman | Cassandra: a Decentralized Structured Storage System[END_REF]) are initiated in the background. In the case of consistency-level ALL (dots) we observe a measurable degradation of about 25% in the following minute and immediate recovery of service after that. This happens because some writes cannot get acknowledgments from the failed node and thus temporarily block until the failover mechanism has been activated. In all cases, CassMail can rely on Cassandra to gracefully handle the node failure with minimal or no availability loss, and without operator intervention.

Comparing CassMail experimentally to systems with equivalent functionality is hard since to the best of our knowledge no such systems are available in open source. The closest alternative -lacking several of CassMail's propertieswould be a system relying on static partitioning of users over conventional SMTP servers. For the purpose of comparing to such a system we configured its build- ing block (an SMTP server based on Postfix 2.7.1) and used it as a reference point for comparing CassMail's single-node performance with a mature tighlyconfigured software system. We did not focus on larger-scale experiments since with static partitioning, client awareness of data location, and no replication one can trivially achieve linear scalability up to the limits of the network. We used the Postal benchmark configured as described earlier and created a mailbox file for each user in the server's file system. Our results show that the Postfixbased server achieves average write throughput of 40MB/s and 25MB/s with large (200KB-2MB) and small (50KB-500KB) messages respectively, limited by CPU in both cases. This contrasts to CassMail's single-node write performance of 15MB/s and 10MB/s for large and small messages respectively. The performance difference can be attributed to implementation characteristics: CassMail is written in high-level programming languages and libraries (Python, Java) and combines e-mail protocol processing with significant storage system processing at each server node. Postfix on the other hand, is a mature performance-optimized software system written in C using a lightweight storage stack. We believe that CassMail's scalability properties can make up for the impact in single-node performance.

6 Discussion and future work

We are exploring deployment of CassMail in a Cloud infrastructure [START_REF] Armbrust | Above the Clouds: A Berkeley View of Cloud Computing[END_REF][START_REF] Elson | Handling Flash Crowds from your Garage[END_REF] offering virtual machines (VMs) and local or remotely-mounted storage volumes. In a straightforward deployment scheme each Cassandra server maps to a VM and each disk to (possibly RAID setups of) local or remotely-mounted storage volumes. Assumptions about failure independence require VMs and storage volumes to not share any single point of failure (such as a physical server). Current

Cloud providers hide this level of information from the user raising a challenge to effective deployment. Our experimental results suggest the use of VMs with considerable CPU (number of cores) and physical memory allocations. In addition, the higher performance, reliability, and predictability of local storage makes it a better alternative to remotely-mounted storage for storing Cassandra data.

Our system has proven to be quite robust under intensive experiments but is currently lacking some features that are needed for real-world deployment. First, it does not deal with user authentication or data encryption of the messages being transferred. Also, the SMTP server currently receives e-mail but does not relay messages to other mailservers. We believe that these features are straightforward additions to our prototype and we plan to implement them in the near future.

Conclusions

Eventually-consistent storage systems have been shown to be key enablers of scalable, highly available, and manageable application services that do not require strict consistency semantics. In particular, Porcupine [START_REF] Saito | Manageability, Availability, and Performance in Porcupine: A Highly Scalable, Cluster-based Mail Service[END_REF] demonstrated a scalable e-mail service that was based on an eventually-consistent storage system built from scratch. A drawback of such an approach is the complexity and long development effort it requires (Porcupine consists of 14 major components written in C++ with a total of about 41,000 lines of code [START_REF] Saito | Manageability, Availability, and Performance in Porcupine: A Highly Scalable, Cluster-based Mail Service[END_REF]). The emergence of general-purpose scalable storage services that offer APIs with eventualconsistency semantics such as Cassandra raise the opportunity of realizing application services similar to Porcupine at a lower development cost.

In this paper we describe the results of a project to build an SMTP/POP service over Cassandra and show that such a service can be simple (consisting of a few tens of lines of Python code focused on the application logic) and thus rapidly implemented. We also show that the implementation exhibits good scalability properties: its throughput increases from 15MB/s to 55MB/s when the cluster size grows from 1 to 8 nodes and a crash failure of a single node results in minimal to no availability lapse, depending on the level of consistency. These properties also indicate an easy-to-manage system (no need for human intervention in mapping users to storage nodes or for restoring availability at the time of failure), which is a critical characteristic of a system meant to operate at a large scale.

A cost for the simplicity of our design is the additional overhead (evidenced by high CPU usage of Python SMTP/POP server and Java Cassandra client/servers) as well as the background activity inherent in scripted and garbage-collected programming environments. However, the combination of technology trends pointing to more cycles in future multi-core CPUs (which can to some extent absorb higher overheads of high-level language runtimes) and the strength of the eventually-consistent storage model in hiding the effect of slow replicas in write performance, paint a positive conclusion: We believe that the synthesis of interoperable (application and storage) components is a viable path to rapidly prototyping robust scalable systems. In this context, storage systems with general-purpose APIs that explore alternative consistency semantics are important foundational abstractions for building scalable applications.

Fig. 1 .

 1 Fig. 1. The Cassandra schema designed for CassMail.

Fig. 2 .

 2 Fig. 2. System design

Fig. 3 .

 3 Fig. 3. Throughput for different replication factors.

Fig. 4 .

 4 Fig. 4. Throughput for different consistency levels.

Fig. 5 .

 5 Fig. 5. Throughput for different message size ranges.

Fig. 6 .

 6 Fig.6. Throughput over time before and after the occurrence of a failure event.

Table 1 .

 1 Protocol commands supported by SMTP and POP3 servers.

	Server	Supported commands
	SMTP	HELO MAIL RCPT DATA RSET NOOP QUIT
	POP3 USER PASS STAT LIST UIDL TOP RETR DELE RSET NOOP QUIT

Normally an e-mail goes through the submission, relaying, and delivery steps. Without loss of generality in this discussion we omit relaying and think of the users connecting directly to a system node to submit and then deliver the message.

http://www.ruby-doc.org/stdlib/libdoc/gserver/rdoc/

https://github.com/fauna/cassandra

https://github.com/pycassa/pycassa

http://incubator.apache.org/thrift/

http://wiki.apache.org/cassandra/StorageConfiguration

http://cassandra.apache.org/

http://doc.coker.com.au/projects/postal/

Acknowledgments

We thankfully acknowledge the support of the European ICT-FP7 program through the SCALEWORKS (MC IEF 237677) and CUMULONIMBO (STREP 257993) projects.