Search

Many geo-distributed data analytics (GDA) systems have focused on the network performance-bottleneck: inter-data center network bandwidth to improve performance. Unfortunately, these systems may encounter a cost-bottleneck ( ${\$}$ $ ) because they have not considered data transfer cost ( ${\$}$ $ ), one of the most expensive and heterogeneous resources in a multi-cloud environment. In this article, we present Kimchi , a network cost-aware GDA system to meet the cost-performance tradeoff by exploiting data transfer cost heterogeneity to avoid the cost-bottleneck. Kimchi determines cost-aware task placement decisions for scheduling tasks given inputs including data transfer cost, network bandwidth, input data size and locations, and desired cost-performance tradeoff preference. In addition, Kimchi is also mindful of data transfer cost in the presence of dynamics. Kimchi has been applied to two common GDA MapReduce models: synchronous barrier and asynchronous push-based shuffle. A Kimchi prototype has been implemented on Spark, and experiments show that it reduces cost by 5% $\scriptstyle \sim$ ∼ 24% without impacting performance and reduces query execution time by 45% $\scriptstyle \sim$ ∼ 70% without impacting cost compared to other baseline approaches centralized, vanilla Spark, and bandwidth-aware (e.g., Iridium). More importantly, Kimchi allows applications to explore a much richer cost-performance tradeoff space in a multi-cloud environment.

Multi-tiered geo-distributed cloud storage systems must tame complexity at many levels: uniform APIs for storage access, supporting flexible storage policies that meet a wide array of application metrics, determining an optimal data placement, handling uncertain network dynamics and access dynamism, and operating across many levels of heterogeneity both within and across data-centers (DCs). In this paper, we present an integrated solution called Wiera. Wiera enables the specification of data management policies both within a local DC and across DCs. Such policies enable the user to optimize for cost, performance, reliability, durability, and consistency, and to express their tradeoffs. In addition, Wiera determines an optimal data placement for the user to meet their desired tradeoffs easily in such an environment. A key aspect of Wiera is first-class support for dynamism due to network, workload, and access patterns changes. As far as we know, Wiera is the first geo-distributed cloud storage system which handles dynamism actively at run-time. Wiera allows unmodified applications to reap the benefits of flexible data/storage policies by externalizing the policy specification. We show how Wiera enables a rich specification of dynamic policies using a concise notation and describe the design and implementation of the system. We have implemented a Wiera prototype on multiple cloud environments, AWS and Azure, that illustrates potential benefits from managing dynamics and in using multiple cloud storage tiers both within and across DCs.

Adaptability is critical for stream processing systems to ensure stable, low-latency, and high-throughput processing of long-running queries. Such adaptability is particularly challenging for wide-area stream processing due to the highly dynamic nature of the wide-area environment, which includes unpredictable workload patterns, variable network bandwidth, occurrence of stragglers, and failures. Unfortunately, existing adaptation techniques typically achieve these performance goals by compromising the quality/accuracy of the results, and they are often application-dependent. In this work, we rethink the adaptability property of wide-area stream processing systems and propose a resource-aware adaptation framework, called WASP. WASP adapts queries through a combination of multiple techniques: task re-assignment, operator scaling, and query re-planning, and applies them in a WAN-aware manner. It is able to automatically determine which adaptation action to take depending on the type of queries, dynamics, and optimization goals. We have implemented a WASP prototype on Apache Flink. Experimental evaluation with the YSB benchmark and a real Twitter trace shows that WASP can handle various dynamics without compromising the quality of the results.

Edge environments are generating an increasingly large amount of data due to the proliferation of edge devices. Accommodating this large influx of data at edge servers is a challenging issue. While some data can be processed as it is generated, others must be stored for later access. This paper proposes the features that a new edge-native storage system must possess including support for user mobility and node fluctuation. To motivate this, we first describe several emerging edge applications and their data needs. We then describe the challenges in meeting these needs. We then evaluate an out-of-the-box cloud storage system, Cassandra, to assess it’s suitability as an edge storage system due to many edge-friendly features. We determined that while a cloud-based storage system can be ported to the edge meeting some of the challenges, other challenges require new solutions. Based on the challenges and the results of Cassandra case study, we propose a set of design principles for a new edge-native storage system.

Many geo-distributed data analytics (GDA) systems have focused on the network performance-bottleneck: interdata center network bandwidth to improve performance. Unfortunately, these systems may encounter a cost-bottleneck ($) because they have not considered data transfer cost ($), one of the most expensive and heterogeneous resources in a multi-cloud environment. In this paper, we present Kimchi, a network cost-aware GDA system to meet the cost-performance tradeoff by exploiting data transfer cost heterogeneity to avoid the cost-bottleneck. Kimchi determines cost-aware task placement decisions for scheduling tasks given inputs including data transfer cost, network bandwidth, input data size and locations, and desired cost-performance tradeoff preference. In addition, Kim- chi is also mindful of data transfer cost in the presence of dynamics. A Kimchi prototype has been implemented on Spark and experiments show that it reduces cost by 14% ~ 24% without impacting performance and reduces query execution time by 45% ~ 70% without impacting cost compared to other baseline approaches centralized, vanilla Spark, and bandwidth-aware (e.g. Iridium). More importantly, Kimchi allows applications to explore a much richer cost-performance tradeoff space in a multi-cloud environment.

Centralized cloud infrastructures have become the popular platforms for data-intensive computing today. However, they suffer from inefficient data mobility due to the centralization of cloud resources, and hence, are highly unsuited for geo-distributed data-intensive applications where the data may be spread at multiple geographical locations. In this paper, we present Nebula: a dispersed edge cloud infrastructure that explores the use of voluntary resources for both computation and data storage. We describe the lightweight Nebula architecture that enables distributed data-intensive computing through a number of optimization techniques including location-aware data and computation placement, replication, and recovery. We evaluate Nebula performance on an emulated volunteer platform that spans over 50 PlanetLab nodes distributed across Europe, and show how a common data-intensive computing framework, MapReduce, can be easily deployed and run on Nebula. We show Nebula MapReduce is robust to a wide array of failures and substantially outperforms other wide-area versions based on emulated existing systems.

MapReduce has proven remarkably effective for a wide variety of data-intensive applications, but it was designed to run on large single-site homogeneous clusters. Researchers have begun to explore the extent to which the original MapReduce assumptions can be relaxed, including skewed workloads, iterative applications, and heterogeneous computing environments. This paper continues this exploration by applying MapReduce across geo-distributed data over geo-distributed computation resources. Using Hadoop, we show that network and node heterogeneity and the lack of data locality lead to poor performance, because the interaction of MapReduce phases becomes pronounced in the presence of heterogeneous network behavior. To address these problems, we take a two-pronged approach: We first develop a model-driven optimization that serves as an oracle, providing high-level insights. We then apply these insights to design cross-phase optimization techniques that we implement and demonstrate in a real-world MapReduce implementation. Experimental results in both Amazon EC2 and PlanetLab show the potential of these techniques as performance is improved by 7-18 percent depending on the execution environment and application.

In the era of rapid experimental expansion data analysis needs are rapidly outpacing the capabilities of small institutional clusters and looking to integrate HPC resources into their workflow. We propose one way of reconciling on-demand needs of experimental analytics with the batch managed HPC resources within a system that dynamically moves nodes between an on-demand cluster configured with cloud technology (OpenStack) and a traditional HPC cluster managed by a batch scheduler (Torque). We evaluate this system experimentally both in the context of real-life traces representing two years of a specific institutional need, and via experiments in the context of synthetic traces that capture generalized characteristics of potential batch and on-demand workloads. Our results for the real-life scenario show that our approach could reduce the current investment in on-demand infrastructure by 82% while at the same time improving the mean batch wait time almost by an order of magnitude (8x).

Wide-area data analytics has gained much attention in recent years due to the increasing need for analyzing data that are geographically distributed. Many of such queries often require real-time analysis on data streams that are continuously being generated across multiple locations. Yet, analyzing these geo-distributed data streams in a timely manner is very challenging due to the highly heterogeneous and limited bandwidth availability of the wide-area network (WAN). This paper examines the opportunity of applying multi-query optimization in the context of wide-area streaming analytics, with the goal of utilizing WAN bandwidth efficiently while achieving high throughput and low latency execution. Our approach is based on the insight that many streaming analytics queries often exhibit common executions, whether in consuming a common set of input data or performing the same data processing. In this work, we study different types of sharing opportunities and propose a practical online algorithm that allows streaming analytics queries to share their common executions incrementally. We further address the importance of WAN awareness in applying multi-query optimization. Without WAN awareness, sharing executions in a wide-area environment may lead to performance degradation. We have implemented our WAN-aware multi-query optimization in a prototype implementation based on Apache Flink. Experimental evaluation using Twitter traces on a real wide-area system deployment across geo-distributed EC2 data centers shows that our technique is able to achieve 21% higher throughput while saving WAN bandwidth consumption by 33% compared to a WAN-aware, sharing-agnostic system.

The past few years have seen a growing number of mobile and sensor applications that rely on Cloud support. The role of the Cloud is to allow these resource-limited devices to offload and execute some of their compute-intensive tasks in the Cloud for energy saving and/or faster processing. However, such offloading to the Cloud may result in high network overhead which is not suitable for many mobile/sensor applications that require low latency. So, people have looked at an alternative Cloud design whose resources are located at the edge of the Internet, called Edge Cloud. Although the use of Edge Cloud can mitigate the offloading overhead, the computational power and network bandwidth of Edge Cloud's resources are typically much more limited compared to the centralized Cloud and hence are more sensitive to workload variation (e.g., due to CPU or I/O contention). In this paper, we propose a locality-aware load sharing technique that allows edge resources to share their workload in order to maintain the low latency requirement of Mobile-Cloud applications. Specifically, we study how to determine which edge nodes should be used to share the workload with and how much of the workload should be shared to each node. Our experiments show that our locality-aware load sharing technique is able to maintain low average end-to-end latency of mobile applications with low latency variation, while achieving good utilization of resources in the presence of a dynamic workload.

Centralized cloud platforms have been widely utilized for data-intensive computing in many domains. However, such systems are not suitable for geo-distributed applications since they require data to be moved to a central location for processing. Recent works have proposed an alternative cloud platform, called edge cloud that provides computational and/or storage resources at the edge, enabling in-situ data processing and low latency. Although such a dispersed cloud model offers low latency, it comes with reliability trade-offs. First, edge resources are interconnected using a wide-area network which is less reliable compared to an intra-cluster network. Second, resources in the edge cloud are typically highly heterogeneous leading to performance variability. Third, edge resources may span different organizational domains, containing different participation rules, leading to greater unreliability. In this paper, we discuss the issues of reliable computation and data storage availability in a geodistributed edge cloud system built using commodity resources. We introduce a notion of reliability factor which defines how reliable a node is. Using this reliability factor, we schedule tasks to a set of nodes to meet a certain reliability goal and dynamically replicate data to achieve timeliness for computation and high data availability for data storage respectively. We evaluate our techniques on the Nebula edge cloud and find that the use of reliability factor results in better performance and storage utilization.

Exploiting the cloud storage hierarchy both within and across data-centers of different cloud providers empowers Internet applications to choose data centers (DCs) and storage services based on storage needs. However, using multiple storage services across multiple data centers brings a complex data placement problem that depends on a large number of factors including, e.g., desired goals, storage and network characteristics, and pricing policies. In addition, dynamics e.g., changing user locations and access patterns, make it impossible to determine the best data placement statically. In this paper, we present TripS, a lightweight system that considers both data center locations and storage tiers to determine the data placement for geo-distributed storage systems. Such systems make use of TripS by providing inputs including SLA, consistency model, fault tolerance, latency information, and cost information. With given inputs, TripS models and solves the data placement problem using mixed integer linear programming (MILP) to determine data placement. In addition, to adapt quickly to dynamics, we introduce the notion of Target Locale List (TLL), a pro-active approach to avoid expensive re-evaluation of the optimal placement. The TripS prototype is running on Wiera, a policy driven geo-distributed storage system, to show how a storage system can easily utilize TripS for data placement. We evaluate TripS/Wiera on multiple data centers of AWS and Azure. The results show that TripS/Wiera can reduce cost 14.96% ∼ 98.1% based on workloads in comparison with other works' approaches and can handle both short- and long-term dynamics to avoid SLA violations.

The Internet of Things (IoT) is enabling a multitude of new applications. Current systems exhibit inefficiencies which stem from applications being isolated and opportunities for sharing and contention resolution ignored. We propose a new system called Constellation that enables synergies between applications which we believe can make IoT systems more efficient. We explore such synergies and provide preliminary results to legitimize our approach.

Interest has been increasing in mining information from a huge amount of data for various purposes such as business processes, scientific discoveries, and security. MINDS (Minnesota INtrusion Detection System) is a tool utilizing data mining techniques to identify both known and unknown network intrusions. While performing well with a high degree of accuracy in detecting intrusions, MINDS is intrinsically limited due to its centralized design. At the same time, there are increasing needs for coordinating multiple distributed sites to find correlations of suspicious events and for sharing distributed anomaly data in different administrative domains since network intrusions become more sophisticated and harder to identify by a single site analysis. In this work, we consider a distributed model of MINDS to eliminate the limitations that the current MINDS suffers from due to its centralized nature. We utilize Grid technologies to achieve the distribution of services. In addition, we develop security frameworks for authentication and access control since it is important to preserve privacy and sensitivity of data regardless of distribution. To verify the feasibility of our approach, we also present our experimental results conducted in a local testbed with four nodes and in a wide-area setting in PlanetLab with 20 nodes geographically dispersed.

Cloud computing services are traditionally deployed on centralized computing infrastructures confined to a few data centers, while cloud applications run in a single data center. However, the cloud's centralized nature can be limiting in terms of performance and cost for applications where users, data, and computation are distributed. The authors present an overview of distributed clouds that might be better suited for such applications. They briefly describe the distributed cloud landscape and introduce Nebula, a highly decentralized cloud that uses volunteer edge resources. The authors provide insights into some of its key properties and design issues, and describe a distributed MapReduce application scenario to illustrate the benefits and trade-offs of using distributed and decentralized clouds for distributed data-intensive computing applications.

Geo-distributed cloud storage systems must tame complexity at many levels: uniform APIs for storage access, supporting flexible storage policies that meet a wide array of application metrics, handling uncertain network dynamics and access dynamism, and operating across many levels of heterogeneity both within and across data-centers. In this paper, we present an integrated solution called Wiera. Wiera extends our earlier cloud storage system, Tiera, that is targeted to multi-tiered policy-based single cloud storage, to the wide-area and multiple data-centers (even across different providers). Wiera enables the specification of global data management policies built on top of local Tiera policies. Such policies enable the user to optimize for cost, performance, reliability, durability, and consistency, both within and across data-centers, and to express their tradeoffs. A key aspect of Wiera is first-class support for dynamism due to network, workload, and access patterns changes. Wiera policies can adapt to changes in user workload, poorly performing data tiers, failures, and changes in user metrics (e.g., cost). Wiera allows unmodified applications to reap the benefits of flexible data/storage policies by externalizing the policy specification. As far as we know, Wiera is the first geo-distributed cloud storage system which handles dynamism actively at run-time. We show how Wiera enables a rich specification of dynamic policies using a concise notation and describe the design and implementation of the system. We have implemented a Wiera prototype on multiple cloud environments, AWS and Azure, that illustrates potential benefits from managing dynamics and in using multiple cloud storage tiers both within and across data-centers.

Today, many organizations need to operate on datathat is distributed around the globe. This is inevitable due to thenature of data that is generated in different locations such as videofeeds from distributed cameras, log files from distributed servers, and many others. Although centralized cloud platforms havebeen widely used for data-intensive applications, such systemsare not suitable for processing geo-distributed data due to highdata transfer overheads. An alternative approach is to use anEdge Cloud which reduces the network cost of transferringdata by distributing its computations globally. While the EdgeCloud is attractive for geo-distributed data-intensive applications, extending existing cluster computing frameworks to a wide-areaenvironment must account for locality. We propose Awan: anew locality-aware resource manager for geo-distributed dataintensiveapplications. Awan allows resource sharing betweenmultiple computing frameworks while enabling high localityscheduling within each framework. Our experiments with theNebula Edge Cloud on PlanetLab show that Awan achieves up toa 28% increase in locality scheduling which reduces the averagejob turnaround time by approximately 18% compared to existingcluster management mechanisms.

It is generally accepted that the ability to develop large-scale distributed applications has lagged seriously behind other developments in cyberinfrastructure. In this paper, we provide insight into how such applications have been developed and an understanding of why developing applications for distributed infrastructure is hard. Our approach is unique in the sense that it is centered around half a dozen existing scientific applications; we posit that these scientific applications are representative of the characteristics, requirements, as well as the challenges of the bulk of current distributed applications on production cyberinfrastructure (such as the US TeraGrid). We provide a novel and comprehensive analysis of such distributed scientific applications. Specifically, we survey existing models and methods for large-scale distributed applications and identify commonalities, recurring structures, patterns and abstractions. We find that there are many ad hoc solutions employed to develop and execute distributed applications, which result in a lack of generality and the inability of distributed applications to be extensible and independent of infrastructure details. In our analysis, we introduce the notion of application vectors: a novel way of understanding the structure of distributed applications. Important contributions of this paper include identifying patterns that are derived from a wide range of real distributed applications, as well as an integrated approach to analyzing applications, programming systems and patterns, resulting in the ability to provide a critical assessment of the current practice of developing, deploying and executing distributed applications. Gaps and omissions in the state of the art are identified, and directions for future research are outlined.

As the number of services and the size of data involved in workflows increases, centralized orchestration techniques are reaching the limits of scalability. When relying on web services without third-party data transfer, a standard orchestration model needs to pass all data through a centralized engine, which results in unnecessary data transfer and the engine to become a bottleneck to the execution of a workflow. As a solution, this paper presents and evaluates Circulate, an alternative service-oriented architecture which facilitates an orchestration model of central control in combination with a choreography model of optimized distributed data transport. Extensive performance analysis through the PlanetLab framework is conducted on a web service-based implementation over a range of Internet-scale configurations which mirror scientific workflow environments. Performance analysis concludes that our architecture's optimized model of data transport speeds up the execution time of workflows, consistently outperforms standard orchestration and scales with data and node size. Furthermore, Circulate is a less-intrusive solution as individual services do not have to be reconfigured in order to take part in a workflow.

Distributed computing applications are increasingly utilizing distributed data sources. However, the unpredictable cost of data access in large-scale computing infrastructures can lead to severe performance bottlenecks. Providing predictability in data access is, thus, essential to accommodate the large set of newly emerging large-scale, data-intensive computing applications. In this regard, accurate estimation of network performance is crucial to meeting the performance goals of such applications. Passive estimation based on past measurements is attractive for its relatively small overhead compared to relying on explicit probing. In this paper, we take a passive approach for network performance estimation. Our approach is different from existing passive techniques that rely either on past direct measurements of pairs of nodes or on topological similarities. Instead, we exploit secondhand measurements collected by other nodes without any topological restrictions. In this paper, we present Overlay Passive Estimation of Network performance (OPEN), a scalable framework providing end-to-end network performance estimation based on secondhand measurements, and discuss how OPEN achieves cost-effective estimation in a large-scale infrastructure. Our extensive experimental results show that OPEN estimation can be applicable for replica and resource selections commonly used in distributed computing.

Scheduling multisource divisible loads is a challenging task as different sources should cooperate and share their computing power with others to balance their loads and minimize total computational time. In this study, we attempt to address a generalized divisible load scheduling problem for handling loads from multiple sources on arbitrary networks. This problem is all the more challenging as 1) the topology is arbitrary, 2) in such networks, it is difficult to decide from which source and which route a processing node should receive loads, and 3) processing nodes must be allocated to different sources when they become available. We study two distinct cases of interest, static case and dynamic case, and propose two novel strategies, referred to as static scheduling strategy (SSS) and dynamic scheduling strategy (DSS), respectively. Both strategies work in an iterative fashion. In each iteration, they will use a novel graph partitioning (GP) scheme to partition the network such that each source in the network gains a portion of network resources and then these sources cooperate to process their loads. We analyze the performance of DSS using queuing theory and derive upper bounds on a load's average waiting time and a source's average queue length. We use simulation to verify the usefulness and effectiveness of SSS and DSS. Our findings reveal an interesting ?load insensitive? propertyof SSS and also verify the theoretical upper bound of average queue length at each source in the dynamic case.

Large-scale distributed systems provide an attractive scalable infrastructure for network applications. However, the loosely coupled nature of this environment can make data access unpredictable, and in the limit, unavailable. We introduce the notion of accessibility to capture both availability and performance. An increasing number of data-intensive applications require not only considerations of node computation power but also accessibility for adequate job allocations. For instance, selecting a node with intolerably slow connections can offset any benefit to running on a fast node. In this paper, we present accessibility-aware resource selection techniques by which it is possible to choose nodes that will have efficient data access to remote data sources. We show that the local data access observations collected from a node's neighbors are sufficient to characterize accessibility for that node. By conducting trace-based, synthetic experiments on PlanetLab, we show that the resource selection heuristics guided by this principle significantly outperform conventional techniques such as latency-based or random allocations. The suggested techniques are also shown to be stable even under churn despite the loss of prior observations.

Network service-based computation is a promising paradigm for both scientific and engineering, and enterprise computing. The network service allows users to focus on their application and obtain services when needed, simply by invoking the service across the network. In this paper, we show that an adaptive, general-purpose run-time infrastructure in support of effective resource management can be built for a wide range of high-end network services running in a single-site cluster and in a Grid. The primary components of the run-time infrastructure are: (1) dynamic performance prediction; (2) adaptive intra-site resource management; and (3) adaptive inter-site resource management. The novel aspect of our approach is that the run-time system is able to dynamically select the most appropriate performance predictor or resource management strategy over time. This capability not only improves the performance, but also makes the infrastructure reusable across different high-end services. To evaluate the effectiveness and applicability of our approach, we have transformed two different classes of high-end applications-data parallel and distributed applications-into network services using the infrastructure. The experimental results show that the network services running on the infrastructure significantly reduce the overall service times under dynamically varying circumstances.

As the number of services and the size of data involved in workflows increases, centralised orchestration techniques are reaching the limits of scalability. In the classic orchestration model, all data passes through a centralised engine, which results in unnecessary data transfer, wasted bandwidth and the engine to become a bottleneck to the execution of a workflow. This paper presents and evaluates the Circulate architecture which maintains the robustness and simplicity of centralised orchestration, but facilitates choreography by allowing services to exchange data directly with one another. Circulate could be realised within any existing workflow framework, in this paper, we focus on WS-Circulate, a Web services based implementation. Taking inspiration from the Montage workflow, a number of common workflow patterns (sequence, fan-in and fan-out), input to output data size relationships and network configurations are identified and evaluated. The performance analysis concludes that a substantial reduction in communication overhead results in a 2---4 fold performance benefit across all patterns. An end-to-end pattern through the Montage workflow results in an 8 fold performance benefit and demonstrates how the advantage of using the Circulate architecture increases as the complexity of a workflow grows.

This paper addresses the inherent unreliability and instability of worker nodes in large-scale donation-based distributed infrastructures such as peer-to-peer and grid systems. We present adaptive scheduling techniques that can mitigate this uncertainty and significantly outperform current approaches. In this work, we consider nodes that execute tasks via donated computational resources and may behave erratically or maliciously. We present a model in which reliability is not a binary property, but a statistical one based on a node's prior performance and behavior. We use this model to construct several reputation-based scheduling algorithms that employ estimated reliability ratings of worker nodes for efficient task allocation. Our scheduling algorithms are designed to adapt to changing system conditions, as well as nonstationary node reliability. Through simulation, we demonstrate that our algorithms can significantly improve throughput while maintaining a very high success rate of task completion. Our results suggest that reputation-based scheduling can handle a wide variety of worker populations, including nonstationary behavior, with overhead that scales well with system size. We also show that our adaptation mechanism allows the application designer fine-grain control over the desired performance metrics.

Large-scale distributed applications are subject to frequent disruptions due to resource contention and failure. Such disruptions are inherently unpredictable and, therefore, robustness is a desirable property for the distributed operating environment. In this work, we describe and evaluate a robust topology for applications that operate on a spanning tree overlay network. Unlike previous work that is adaptive or reactive in nature, we take a proactive approach to robustness. The topology itself is able to simultaneously withstand disturbances and exhibit good performance. We present both centralized and distributed algorithms to construct the topology, and then demonstrate its effectiveness through analysis and simulation of two classes of distributed applications: Data collection in sensor networks and data dissemination in divisible load scheduling. The results show that our robust spanning trees achieve a desirable trade-off for two opposing metrics where traditional forms of spanning trees do not. In particular, the trees generated by our algorithms exhibit both resilience to data loss and low power consumption for sensor networks. When used as the overlay network for divisible load scheduling, they display both robustness to link congestion and low values for the makespan of the schedule

In today's batch queue HPC cluster systems, the user submits a job requesting a fixed number of processors. The system will not start the job until all of the requested resources become available simultaneously. When cluster workload is high, large sized jobs will experience long waiting time due to this policy. In this paper, we propose a new approach that dynamically decomposes a large job into smaller ones to reduce waiting time, and lets the application expand across multiple subjobs while continuously achieving progress. This approach has three benefits: (i) application turnaround time is reduced, (ii) system fragmentation is diminished, and (iii) fairness is promoted. Our approach does not depend on job queue time prediction but exploits available backfill opportunities. Simulation results have shown that our approach can reduce application mean turnaround time by up to 48%.

Many Cloud applications exploit the diversity of storage options in a data center to achieve desired cost, performance, and durability tradeoffs. It is common to see applications using a combination of memory, local disk, and archival storage tiers within a single data center to meet their needs. For example, hot data can be kept in memory using ElastiCache, and colder data in cheaper, slower storage such as S3, using Amazon as an example. For user-facing applications, a recent trend is to exploit multiple data centers for data placement to enable better latency of access from users to their data. The conventional wisdom is that co-location of computation and storage within the same data center is a key to application performance, so that applications running within a data center are often still limited to access local data. In this paper, using experiments on Amazon, Microsoft, and Google clouds, we show that this assumption is false, and that accessing data in nearby data centers may be faster than local access at different or even same points in the storage hierarchy. This can lead to not only better performance, but also reduced cost, simpler consistency policies and reconsidering data locality in multiple DCs environment. This argues for an expansion of cloud storage tiers to consider non-local storage options, and has interesting implications for the design of a distributed storage system.

One of the factors that limits the scale, performance, and sophistication of distributed applications is the difficulty of concurrently executing them on multiple distributed computing resources. In part, this is due to a poor understanding of the general properties and performance of the coupling between applications and dynamic resources. This paper addresses this issue by integrating abstractions representing distributed applications, resources, and execution processes into a pilot-based middleware. The middleware provides a platform that can specify distributed applications, execute them on multiple resource and for different configurations, and is instrumented to support investigative analysis. We analyzed the execution of distributed applications using experiments that measure the benefits of using multiple resources, the late-binding of scheduling decisions, and the use of backfill scheduling.

The abundance of compute and storage resources available in the cloud makes it well-suited to addressing the limitations of mobile devices. We explore the use of cloud infrastructure to optimize content-centric mobile applications, which can have high communication and storage requirements, based on the analysis of user activity. We present two specific optimizations, precaching and prefetching, as well as the design and implementation of a middleware framework that allows mobile application developers to easily utilize these techniques. Our framework is fully generalizable to any content-centric mobile application, a large and growing class of Internet applications. A news aggregation application is used as a case study to evaluate our implementation. We make use of a cosine similarity scheme to identify users with similar interests, which in turn is used to determine what content to prefetch. Various cache algorithms, implemented for our framework, are also considered. A workload trace and simulation are used to measure the performance of the application and framework. We observe a dramatic improvement in application performance due to use of our framework with a reasonable amount of overhead. Our system also significantly outperforms a baseline implementation that performs the same optimizations without taking user activity into account.

In this paper, we explore the trade-offs and opportunities in porting a high-throughput Grid computing middleware to a high-performance service oriented environment. We present the limitations of the Grid computing middleware when operating in such a performance sensitive enviroment and suggest ways of overcoming these limitations. We focus on exploiting the computation and communiction heterogeneity of the Grid resources to meet the peformance requirements of services, and present several approaches of work distribution that deal with this hetergeneity. We also present a heuristic for finding the best decomposition of work and present algorithms for each of the approaches which we evaluate on a PlanetLab tesbed. The results validate the heuristic and indicate that a significant improvement in performance can be achieved by making the Grid computing middleware aware of the heterogeneity in the underlying infrastructure. The results also provide some useful insights into selecting a work distribution policy based on the dynamic status of the Grid computing environment.

This paper presents a new paradigm for parallel job scheduling called integrated scheduling or iScheduling. The iScheduler is an application-aware job scheduler as opposed to a general-purpose system scheduler. It dynamically controls resource allocation among a set of competing applications, but unlike a traditional job scheduler, it can interact directly with an application during execution to optimize resource allocation. An iScheduler may add or remove resources from a running application to improve the performance of other applications. Such fluid resource management can support both improved application and system performance. We propose a framework for building iSchedulers and evaluate the concept on several workload traces obtained both from supercomputer centers and from a set of real parallel jobs. The results indicate that iScheduling can improve both waiting time and overall turnaround time substantially for these workload classes, outperforming standard policies such as backfilling and moldable job scheduling.

This paper examines the problem of adapting data parallel applications in a shared dynamic environment of PC or workstation clusters. We developed an analytic framework to compare and contrast a wide range of adaptation strategies: dynamic load balancing, migration, processor addition and removal. These strategies have been evaluated with respect to the cost and benefit they provide for three representative parallel applications: an iterative jacobi solver for Laplace's equation, gaussian elimination with partial pivoting, and a gene sequence comparison application. We found that the cost and benefit of each method can be predicted with high accuracy (within 10%) for all applications and show that the framework is applicable to a wide variety of parallel applications. We then show that accurate prediction allows the most appropriate method to be selected dynamically. Performance improvement for the three applications ranged from 25% to 45% using our adaptation library. In addition, we dispel the conventional wisdom that migration is too expensive, and show that it can be beneficial even for running parallel applications with non-trivial communication.

This paper presents the design of a new system architecture, Virtual Service Grid (VSG), for delivering high-performance network services. The VSG is based on the concept of the virtual service which provides location, replication, and fault transparency to clients accessing remotely deployed high-end services. One of the novel features of the virtual service is the ability to self-scale in response to client demand. The VSG exploits network and service information to make adaptive dynamic replica selection, creation, and deletion decisions. We describe the VSG architecture, middleware, and replica management policies. We have deployed the VSG on a wide-area Internet testbed to evaluate its performance. The results indicate that the VSG can deliver efficient performance for a wide range of client workloads, both in terms of reduced response time and in the utilization of system resources. Copyright © 2002 John Wiley & Sons, Ltd.

In this paper, we present our vision for cloud-based mobile computing using user profile information. Such information enables a series of data-driven optimizations: filtering, aggregation, and speculation, that go beyond the well-researched benefit of mobile outsourcing. These optimizations can improve performance, reliability, and energy usage. A novel aspect of our approach is to exploit the unique ability of the cloud to collect and analyze large amounts of user profile data, cache shared data, and even enable sharing of computations, across different mobile users. We present results for two exemplar mobile-cloud applications, driven by workload traces derived from Twitter feeds and Wikipedia document editing, to illustrate these opportunities.

Today, centralized data-centers or clouds have become the de-facto platform for data-intensive computing in the commercial, and increasingly, scientific domains. This is because clouds such as Amazon AWS and Microsoft Azure offer large amounts of monetized co-located computation and storage well suited to typical processing tasks such as batch analytics. However, many Big Data applications rely on data that is geographically distributed, and is not collocated with the centralized computational resources provided by clouds. Examples of such applications include analysis of user data such as blogs, video feeds taken from geographically separated cameras, monitoring and log analysis of server and content distribution network (CDN) logs, and scientific data collected from distributed instruments and sensors. Such applications lead to a number of challenges for efficient data analytics in today's cloud platforms. First, in many applications, data is both large and widely distributed and data upload may constitute a non-trivial portion of the execution time. Second, centralized cloud resources present a single point of failure and network partitions between the data sources and the cloud can also lead to service disruptions. Third, the cost to transport, store, and process data may be outside of the budget of the small-scale application designer or end-user. The paper present Nebula: a dispersed edge cloud infrastructure that provides both computation and data storage to address the above challenges. The use of edge resources is attractive for several reasons. First, there is an increasing amount of computing and storage resources available on the edge, as evidenced by their use in several volunteer computing, filesharing as well as content delivery (CDN) environments. This capacity is likely to increase further with the provision of powerful multi-core, multi-node desktop and home machines coupled with increasing amount of high bandwidth Internet connectivity. Second, edge resources provide locality to data and users naturally, and hence, can be exploited easily for insitu processing. Finally, if cost is an issue, then volunteer edge resources can be utilized at a relatively low cost.

In this paper, we present our vision for data-driven cloud-based mobile computing. We identify the concept of Region of interest (RoI) that reflects the profile of the user in how they access information or interact with applications. Such information enables a series of data-driven optimizations: filtering, aggregation, and speculation, that go beyond the well-researched benefit of mobile outsourcing. These optimizations can improve performance, reliability, and energy usage. A novel aspect of our approach is to exploit the unique ability of the cloud to collect and analyze large amounts of user profile data, cache shared data, and even enable sharing of computations, across different mobile users. We implement two exemplar mobile-cloud applications on an Android/Amazon Elastic Cloud Compute (EC2)-based mobile outsourcing platform, that utilize the RoI abstraction for data-driven optimizations. We presentresults driven by workload traces derived from Twitter feeds and Wikipedia document editing to illustrate the opportunities of using such optimizations.

Cloud providers offer an array of storage services that represent different points along the performance, cost, and durability spectrum. If an application desires the composite benefits of multiple storage tiers, then it must manage the complexity of different interfaces to these storage services and their diverse policies. We believe that it is possible to provide the benefits of customized tiered cloud storage to applications without compromising simplicity using a lightweight middleware. In this paper, we introduce Tiera, a middleware that enables the provision of multi-tiered cloud storage instances that are easy to specify, flexible, and enable a rich array of storage policies and desired metrics to be realized. Tiera's novelty lies in the first-class support for encapsulated tiered cloud storage, ease of programmability of data management policies, and support for runtime replacement and addition of policies and tiers. Tiera enables an application to realize a desired metric (e.g., low latency or low cost) by selecting different storage services that constitute a Tiera instance, and easily specifying a policy, using event and response pairs, to manage the life cycle of data stored in the instance. We illustrate the benefits of Tiera through a prototype implemented on the Amazon cloud. By deploying unmodified MySQL database engine and a TPC-W Web bookstore application on Tiera, we are able to improve their respective throughputs by 47% -- 125% and 46% -- 69%, over standard deployments. We further show the flexibility of Tiera in achieving different desired application metrics with minimal overhead.

This paper presents a paradigm for remote file access called Smart File Objects (SFOs). The SFO is a realization of the ELFS (Extensible File Systems) concept of files as typed objects, but it is applied to wide-area networks (J. Karpovich et al., in “Proceedings of the 9th OOPLSA,” 1994). The SFO is an object-oriented application- specific file access paradigm designed to address the bottleneck imposed by high latency, low bandwidth, unpredictable, and unreliable networks such as the current Internet. Newly emerging network applications such as multimedia, metacomputing, and collaboratories will have different sensitivities to these network “features.” These applications will require a more flexible file access mechanism than what is provided by conventional distributed file systems. The SFO uses application and network information to adaptively prefetch and cache needed data in parallel with the execution of the application to mitigate the impact of the network. Preliminary results indicate that the SFO can provide substantial performance gains for network applications.

Computational grids are an exciting new technology that blends high-performance computing, distributed systems and operating systems. Grids provide a coherent software infrastructure that permits the seamless integration of wide-area resources, such as computers, instruments, devices and data archives, to solve large-scale scientific and engineering problems. In grids, the wide-area network appears to be a large virtual computer. Computational grids are a powerful technology that has engendered a new paradigm for high-performance distributed computing. Teaching about grids is crucial for enabling the future grid workforce and increasing its productivity. Developing high-quality grid courseware is difficult; instructors at the University of Minnesota have provided a helpful Web site based on their experience with teaching a metacomputing class. Grid researchers and educators need to pool their talents and resources to develop effective shareable courseware. To aid this effort, university instructors are developing a repository for grid courseware.

Internet computing has emerged, as an attractive paradigm for many applications that require access to distributed resources such as telemedicine, collaboratories, and transaction systems. For applications that require high performance, the Internet is assumed to be inappropriate. However, it is possible to obtain high performance for parallel scientific applications in an Internet environment. A wide-area scheduling system called Gallop has been developed to exploit opportunities for high performance using remote Internet resources for these applications. This paper describes the Gallop architecture and scheduling model, and performance results obtained for three parallel applications. The initial results indicate that wide-area parallel processing can lead to better performance even with Internet technology, but that current Internet bandwidth is a major bottleneck for file and binary transfer. This overhead limits the class of suitable applications to those that are large-grained or that will be run multiple times to amortize the transfer cost.

Prophet is a run-time scheduling system designed to support the efficient execution of parallel applications written in the Mentat programming language (Grimshaw, 1993). Prior results demonstrated that SPMD applications could be scheduled automatically in an ethernet-based local-area workstation network with good performance (Weissman and Grimshaw, 1994 and 1995). This paper describes our recent efforts to extend Prophet along several dimensions: improved overhead control, greater resource sharing, greater resource heterogeneity, wide-area scheduling, and new application types. We show that both SPMD and task parallel applications can be scheduled effectively in a shared heterogeneous LAN environment containing ethernet and ATM networks by exploiting the application structure and dynamic run-time information.

Distributed data-intensive workflow applications are increasingly relying on and integrating remote resources including community data sources, services, and computational platforms. Increasingly, these are made available as data, SAAS, and IAAS clouds. The execution of distributed data-intensive workflow applications can expose network bottlenecks between clouds that compromise performance. In this paper, we focus on alleviating network bottlenecks by using a proxy network. In particular, we show how proxies can eliminate network bottlenecks by smart routing and perform in-network computations to boost workflow application performance. A novel aspect of our work is the inclusion of multiple proxies to accelerate different workflow stages optimizing different performance metrics. We show that the approach is effective for workflow applications and broadly applicable. Using Montage as an exemplar workflow application, results obtained through experiments on Planet Lab showed how different proxies acting in a variety of roles can accelerate distinct stages of Montage. Our microbenchmarks also show that routing data through select proxies can accelerate network transfer for TCP/UDP bandwidth, delay, and jitter, in general.

Map Reduce has been designed to accommodate large-scale data-intensive workloads running on large single-site homogeneous clusters. Researchers have begun to explore the extent to which the original Map Reduce assumptions can be relaxed including skewed workloads, iterative applications, and heterogeneous computing environments. Our work continues this exploration by applying Map Reduce across widely distributed data over distributed computation resources. This problem arises when datasets are generated at multiple sites as is common in many scientific domains and increasingly e-commerce applications. It also occurs when multi-site resources such as geographically separated data centers are applied to the same Map Reduce job. Using Hadoop, we show that the absence of network and node homogeneity and locality of data lead to poor performance. The problem is that interaction of Map Reduce phases becomes pronounced in the presence of heterogeneous network behavior. In this paper, we propose new cross-phase optimization techniques that enable independent Map Reduce phases to influence one another. We propose techniques that optimize the push and map phases to enable push-map overlap and to allow map behavior to feed back into push dynamics. Similarly, we propose techniques that optimize the map and reduce phases to enable shuffle cost to feed back and affect map scheduling decisions. We evaluate the benefits of our techniques in both Amazon EC2 and Planet Lab. The experimental results show the potential of these techniques as performance is improved from 7%-18% depending on the execution environment and application.

Mentat is an object-oriented parallel processing system designed to simplify the task of writing portable parallel programs for parallel machines and workstation networks. The Mentat compiler and run-time system work together to automatically manage the communication and synchronization between objects. The run-time system marshals member function arguments, schedules objects on processors, and dynamically constructs and executes large-grain data dependence graphs. In this article we present the Mentat run-time system. We focus on three aspects—the software architecture, including the interface to the compiler and the structure and interaction of the principle components of the run-time system; the run-time overhead on a component-by-component basis for two platforms, a Sun SparcStation 2 and an Intel Paragon; and an analysis of the minimum granularity required for application programs to overcome the run-time overhead.