On-site, live video streaming is an important application that many depend on to receive entertainment and local and international news. Streaming live video over wireless (e.g., 3G, LTE, 5G) has begun to replace television production vans that use direct microwave links to the television station since they offer greater physical flexibility and have shorter setup times. In this method, several links are often aggregated together to stream video. Efficiently using multiple links for live, adaptive video streaming is challenging because a protocol must determine the amount of data to send on each link, the video bitrate, and the appropriate level of redundancy so that video frames can be delivered within the user’s strict latency requirements. In this paper, we present Conflux: a multi-homed, adaptive video bitrate protocol for live video streaming. Conflux’s main contribution is the design of a modular live video streaming platform that supports multipath scheduling, video bitrate adaption, and adaptive Forward Error Correction. Conflux presents a probabilistic link quality model that is used in conjunction with a user-specific utility function to determine the video bitrate and redundancy levels that maximize the user’s expected utility. These models are contained in separate modules which serve as building blocks to create customized, multi-homed adaptive video bitrate protocols for users with different requirements. Our experimental results show that Conflux provides more than 22% improvement in video quality over other multi-homed systems for typical, two-link network environments and as much as 108% improvement in more challenging network environments with up to five links.

Classical Paxos-like consensus protocols limit system scalability due to a single leader and the inability to process conflicting proposals in parallel. We introduce a novel agreement protocol, called Antipaxos, that instead reaches agreement on a collection of proposals using an efficient leaderless fast path when the environment is synchronous and failure-free, and falls back on a more elaborate slow path to handle other cases. We first specify the main safety property of Antipaxos by formalizing a new agreement problem called k-Interactive Consistency (k-IC). Then, we present a solution to this problem in the Byzantine failure model. We prove safety and liveness, and also present an experimental performance evaluation in the Amazon cloud. Our experiments show that Antipaxos achieves several-fold higher failure-free peak throughput than Mir-BFT. The inherent efficiency of our approach stems from the low message complexity of the fast path: agreement on n batches of conflict-prone proposals is achieved using only Θ(n2) messages in one consensus cycle, or Θ(n) amortized messages per batch.

This paper evaluates various Machine Learning (ML) techniques for classifying or predicting link instability, which we define as sudden increases in application-level packet loss and latency, or decreases in available bandwidth over a short, pre-determined period of time. To train and evaluate these ML techniques, we first construct attributes from raw wireless LTE traces. These attributes include statistical summaries (e.g., average and standard deviation) and trends (e.g., slope and correlation) of various network metrics. We then determine the attributes that provide the greatest predictive ability to construct our ML examples for training and testing. Using these ML examples, we evaluate the predictive capability of various ML techniques. We find that we can classify 5%-46% of sudden bandwidth drops, 40%-100% of increases in application-level packet loss, and 40%-100% of spikes in latency while maintaining false positive rates below 0.1. In general, we find that Decision Trees and K-Nearest Neighbours techniques provide the highest classification rate while incurring low false positive rates.

This paper introduces Natto, a geo-distributed database system that supports transaction prioritization. Instead of having each shard process transactions in their arrival order, Natto leverages network measurements to estimate the transaction arrival time at each shard, and assigns a timestamp to the transaction based on its arrival time to the furthest shard. These timestamps establish a global ordering of transactions, and introduces opportunities to selectively abort pending low-priority transactions that conflict with a high-priority transaction, or even preempt transactions that are already partially prepared. Our experiments on both Microsoft Azure and a local cluster show that Natto's tail latency for high-priority transactions are significantly lower than the tail latencies of Carousel and TAPIR, which are the current state-of-the-art in geo-distributed transaction processing systems.

This paper introduces Domino, a low-latency state machine replication protocol for wide-area networks. Domino uses network measurements to predict the expected arrival time of a client request to each of its replicas, and assigns a future timestamp to the request indicating when the last replica from the supermajority quorum should have received the request. With accurate arrival time predictions and in the absence of failures, Domino can always commit a request in a single network roundtrip using a Fast Paxos-like protocol by ordering the requests based on their timestamps.

Network traffic typically traverses a sequence of middleboxes forming a service function chain, or simply a chain. Tolerating failures when they occur along chains is imperative to the availability and reliability of enterprise applications. Making a chain fault-tolerant is challenging since, in the event of failures, the state of faulty middleboxes must be correctly and quickly recovered while providing high throughput and low latency.

Sift is a new consensus protocol for replicating state machines. It disaggregates CPU and memory consumption by creating a novel system architecture enabled by one-sided RDMA operations. We show that this system architecture allows us to develop a consensus protocol which centralizes the replication logic. The result is a simplified protocol design with less complex interactions between the participants of the consensus group compared to traditional protocols. The dissaggregated design also enables Sift to reduce deployment costs by sharing backup computational nodes across consensus groups deployed within the same cloud environment. The required storage resources can be further reduced by integrating erasure codes without making significant changes to our protocol. Evaluation results show that in a cloud environment with 100 groups where each group can support up to 2 simultaneous failures, Sift can reduce the cost by 56% compared to an RDMA-based Raft deployment.

In this paper we introduce SpecRPC, a speculative execution framework that allows applications to concurrently execute dependent operations both locally and through remote procedure calls. The framework tracks dependencies among non-speculative and speculative operations and ensures that incorrect speculations do not affect the correctness of applications that follow our suggested design pattern. By using speculation to parallelize dependent operations, SpecRPC can significantly reduce application latency even if only a fraction of the results can be correctly speculated. We evaluate SpecRPC by using it to implement Replicated Commit, a low-latency distributed transaction commit protocol for geo-replicated database systems. Our evaluation results show that, compared to RPC frameworks that sequentially execute dependent operations, SpecRPC can reduce the average transaction completion time of Replicated Commit by 58%.

The trend towards global applications and services has created an increasing demand for transaction processing on globally-distributed data. Many database systems, such as Spanner and CockroachDB, support distributed transactions but require a large number of wide-area network roundtrips to commit each transaction and ensure the transaction's state is durably replicated across multiple datacenters. This can significantly increase transaction completion time, resulting in developers replacing database-level transactions with their own error-prone application-level solutions.

Achieving consensus among a set of distributed entities (or participants) is a fundamental problem at the heart of many distributed systems. A critical problem with most consensus protocols is that they do not scale well. As the number of participants trying to achieve consensus increases, increasing network traffic can quickly overwhelm the network from topology-oblivious broadcasts, or a central coordinator for centralized consensus protocols. Thus, either achieving strong consensus is restricted to a handful of participants, or developers must resort to weaker models of consensus.

Video streaming applications generate a large fraction of Internet traffic. Much of this content is delivered over HTTP using standard web servers. Unlike other types of web workloads, HTTP video streaming workloads are typically disk bound, and therefore an important problem is that of optimizing disk access.

Key-value storage systems are an integral part of many data centre applications, but as demand increases so does the need for high performance. This has motivated new designs that use Remote Direct Memory Access (RDMA) to reduce communication overhead. Current RDMA-enabled key-value stores (RKVSes) target workloads involving small values, running on dedicated servers on which no other applications are running. Outside of these domains, however, there may be other RKVS designs that provide better performance. In this paper, we introduce Nessie, an RKVS that is fully client-driven, meaning no server process is involved in servicing requests. Nessie also decouples its index and storage data structures, allowing indices and data to be placed on different servers. This flexibility can decrease the number of network operations required to service a request. These design elements make Nessie well-suited for a different set of workloads than existing RKVSes. Compared to a server-driven RKVS, Nessie more than doubles system throughput when there is CPU contention on the server, improves throughput by 70 percent for PUT-oriented workloads when data value sizes are 128 KB or larger, and reduces power consumption by 18 percent at 80 percent system utilization and 41 percent at 20 percent system utilization compared with idle power consumption.

In this paper, we introduce Mayflower, a new distributed filesystem that is co-designed from the ground up to work together with a network control plane. In addition to the standard distributed filesystem components, Mayflower has a flow monitor and manager running alongside a software-defined networking controller. This tight coupling with the network controller enables Mayflower to make intelligent replica selection and flow scheduling decisions based on both filesystem and network information. It further enables Mayflower to perform global optimizations that are unavailable to conventional distributed filesystems and network control planes. Our evaluation results from both simulations and a prototype implementation show that Mayflower reduces average read completion time by more than 25% compared to current state-of-the-art distributed filesystems with an independent network flow scheduler, and more than 75% compared to HDFS with ECMP.

A recent paper on benchmarking eventual consistency showed that when a constant workload is applied against Cassandra, the staleness of values returned by read operations exhibits interesting but unexplained variations when plotted against time. In this paper we reproduce this phenomenon and investigate in greater depth the low-level mechanisms that give rise to stale reads. We show that the staleness spikes exhibited by Cassandra are strongly correlated with garbage collection, particularly the "stop-the-world" phase which pauses all application threads in a Java virtual machine. We show experimentally that the staleness spikes can be virtually eliminated by delaying read operations artificially at servers immediately after a garbage collection pause. In our experiments this yields more than a 98% reduction in the number of consistency anomalies that exceed 5ms, and has negligible impact on throughput and latency.

MapReduce has become a popular model for data-intensive computation in recent years. By breaking down each job into small map and reduce tasks and executing them in parallel across a large number of machines, MapReduce can significantly reduce the running time of data-intensive jobs. However, despite recent efforts toward designing resource-efficient MapReduce schedulers, existing solutions that focus on scheduling at the task-level still offer sub-optimal job performance. This is because tasks can have highly varying resource requirements during their lifetime, which makes it difficult for task-level schedulers to effectively utilize available resources to reduce job execution time. To address this limitation, we introduce PRISM, a fine-grained resource-aware MapReduce scheduler that divides tasks into phases, where each phase has a constant resource usage profile, and performs scheduling at the phase level. We first demonstrate the importance of phase-level scheduling by showing the resource usage variability within the lifetime of a task using a wide-range of MapReduce jobs. We then present a phase-level scheduling algorithm that improves execution parallelism and resource utilization without introducing stragglers. In a 10-node Hadoop cluster running standard benchmarks, PRISM offers high resource utilization and provides 1.3× improvement in job running time compared to the current Hadoop schedulers.

This paper proposes a new approach to determining the order of interdependent operations in a distributed system. The key idea behind our approach is to factor the task of tracking happens-before relationships out of components that comprise the system, and to centralize them in a separate event ordering service. This not only simplifies implementation of individual components by freeing them from having to propagate dependence information, but also enables dependence relationships to be maintained across multiple independent systems. A novel API enables the system to detect and take advantage of concurrency whenever possible by maintaining fine-grained information and binding events to a time order as late as possible. We demonstrate the benefits of this approach through several example applications, including a transactional key-value store, and an online graph store. Experiments show that our event ordering service scales well and has low overhead in practice.

Most cloud providers improve resource utilization by having multiple tenants share the same resources. However, this comes at the cost of reduced isolation between tenants, which can lead to inconsistent and unpredictable performance. This performance variability is a significant impediment for tenants running services with strict latency deadlines. Providing predictable performance is particularly important for cloud storage systems. The storage system is the performance bottleneck for many cloud-based services and therefore often determines their overall performance characteristics. In this paper, we introduce MicroFuge, a new distributed caching and scheduling middleware that provides performance isolation for cloud storage systems. MicroFuge addresses the performance isolation problem by building an empirically-driven performance model of the underlying storage system based on measured data. Using this model, MicroFuge reduces deadline misses through adaptive deadline-aware cache eviction, scheduling and load-balancing policies. MicroFuge can also perform early rejection of requests that are unlikely to make their deadlines. Using workloads from the YCSB benchmark on an EC2 deployment, we show that adding MicroFuge to the storage stack substantially reduces the deadline miss rate of a distributed storage system compared to using a deadline oblivious distributed caching middleware such as Memcached.

Most datacenter network (DCN) designs focus on maximizing bisection bandwidth rather than minimizing server-to-server latency. We explore architectural approaches to building low-latency DCNs and introduce Quartz, a design element consisting of a full mesh of switches. Quartz can be used to replace portions of either a hierarchical network or a random network. Our analysis shows that replacing high port-count core switches with Quartz can significantly reduce switching delays, and replacing groups of top-of-rack and aggregation switches with Quartz can significantly reduce congestion-related delays from cross-traffic. We overcome the complexity of wiring a complete mesh using low-cost optical multiplexers that enable us to efficiently implement a logical mesh as a physical ring. We evaluate our performance using both simulations and a small working prototype. Our evaluation results confirm our analysis, and demonstrate that it is possible to build low-latency DCNs using inexpensive commodity elements without significant concessions to cost, scalability, or wiring complexity.

Past work has shown that disk prefetching can be an effective technique for improving the performance of disk bound workloads. However, the performance gains are highly dependent on selecting a prefetch size that is appropriate for a specific system and workload. Using a prefetch size that is too small can lead to poor overall disk throughput, whereas prefetch sizes that are too large can lead to data being evicted before it can be used by a subsequent request.

Event monitoring and logging, that is, recording the communication events between processes, is a critical component in many highly reliable distributed systems. The event logs enable the identification of certain safety-condition violations, such as race conditions and mutual-exclusion violations, as safety is generally contingent on processes communicating in a specific causally ordered pattern. Previous e orts at finding such patterns have often focused on online techniques, which are unable to identify operational problems as they occur. Online monitoring tools exist but they are often restricted to identifying a specific violation condition, such as a deadlock or a race condition, using dedicated data structures. We address the more general problem of detecting causally related event patterns that can be used to identify various undesired behaviours in the system. The main challenge for online pattern matching is the need to store the partial matches to the pattern, as they may combine with future events to form a complete match. Unlike pattern matching in most other domains, causally ordered patterns can span a potentially unbounded number of events and efficiently searching through this large collection poses a significant challenge. In this paper, we introduce OCEP, an efficient online causalevent- pattern-matching framework that bounds the number of partial matches it stores by reporting only a representative subset of pattern matches. We define a subset of matches as representative if it has at least one occurrence of each event in the pattern on each process, which is applicable for a large class of distributed applications. With this definition, OCEP introduces a backtracking algorithm to efficiently find a representative subset from the history of events. An evaluation of the framework shows that OCEP is capable of handling several frequently occurring violation patterns at the event rates of some representative distributed applications.

Large amounts of Internet streaming video traffic are being delivered using HTTP to leverage the existing web infrastructure. A fundamental issue in HTTP streaming concerns the granularity of video objects used throughout the HTTP ecosystem (including clients, proxy caches, CDN nodes, and servers). A video may be divided into many files (called chunks), each containing only a few seconds of video at one extreme, or stored in a single unchunked file at the other.

Distributed key-value stores are now a standard component of high-performance web services and cloud computing applications. While key-value stores offer significant performance and scalability advantages compared to traditional databases, they achieve these properties through a restricted API that limits object retrieval---an object can only be retrieved by the (primary and only) key under which it was inserted. This paper presents HyperDex, a novel distributed key-value store that provides a unique search primitive that enables queries on secondary attributes. The key insight behind HyperDex is the concept of hyperspace hashing in which objects with multiple attributes are mapped into a multidimensional hyperspace. This mapping leads to efficient implementations not only for retrieval by primary key, but also for partially-specified secondary attribute searches and range queries. A novel chaining protocol enables the system to achieve strong consistency, maintain availability and guarantee fault tolerance. An evaluation of the full system shows that HyperDex is 12-13x faster than Cassandra and MongoDB for finding partially specified objects. Additionally, HyperDex achieves 2-4x higher throughput for get/put operations.

Large-scale Internet applications, such as content distribution networks, are deployed across multiple datacenters and consume massive amounts of electricity. To provide uniformly low access latencies, these datacenters are geographically distributed and the deployment size at each location reflects the regional demand for the application. Consequently, an application's environmental impact can vary significantly depending on the geographical distribution of end-users, as electricity cost and carbon footprint per watt is location specific. In this paper, we describe FORTE: Flow Optimization based framework for request-Routing and Traffic Engineering. FORTE dynamically controls the fraction of user traffic directed to each datacenter in response to changes in both request workload and carbon footprint. It allows an operator to navigate the three-way tradeoff between access latency, carbon footprint, and electricity costs and to determine an optimal datacenter upgrade plan in response to increases in traffic load. We use FORTE to show that carbon taxes or credits are impractical in incentivizing carbon output reduction by providers of large-scale Internet applications. However, they can reduce carbon emissions by 10% without increasing the mean latency nor the electricity bill.

Recent increases in live and on-demand video streaming have dramatically changed the Internet landscape. In North America, Netflix alone accounts for 28% of all and 33% of peak downstream Internet traffic on fixed access links, with further rapid growth expected [26]. This increase in streaming traffic coincides with the steady adoption of HTTP for use in video streaming. Many streaming video providers, such as Apple, Adobe, Akamai, Netflix and Microsoft, now use HTTP to stream content [5]. Therefore, it is critical that we understand the impact of this emerging workload on web servers. Unlike other web content, a recent study [13] of streaming video shows that even small infrequent latency spikes, manifested as buffering related pauses, can result in shorter viewing times especially during live broadcasts. Unfortunately, no appropriate benchmarks exist to evaluate web servers under HTTP video streaming workloads.

In this paper, we propose an unorthodox topology for datacenters that eliminates all hierarchical switches in favor of connecting nodes at random according to a small-world-inspired distribution. Specifically, we examine topologies where the underlying nodes are connected at the small scale in a regular pattern, such as a ring, torus or cube, such that every node can route efficiently to nodes in its immediate vicinity, and amended by the addition of random links to nodes throughout the datacenter, such that a greedy algorithm can route packets to far away locations efficiently. Coupled with geographical address assignment, the resulting network can provide content routing in addition to traditional routing, and thus efficiently implement key-value stores. The irregular but self-similar nature of the network facilitates constructing large networks easily using prewired, commodity racks. We show that Small-World Datacenters can achieve higher bandwidth and fault tolerance compared to both conventional hierarchical datacenters as well as the recently proposed CamCube topology. Coupled with hardware acceleration for packet switching, small-world datacenters can achieve an order of magnitude higher bandwidth than a conventional datacenter, depending on the network traffic.

Efficient content distribution in large networks comprising datacenters, end hosts, and distributed in-network caches is a difficult problem. Existing systems rely on mechanisms and metrics that fail to effectively utilize all available sources of bandwidth in the network. This paper presents a novel metric, called the Content Propagation Metric (CPM), for quantitatively evaluating the marginal benefit of available bandwidth to competing consumers, enabling efficient utilization of the bandwidth resource. The metric is simple to implement, imposes only a modest overhead, and can be retrofitted easily into existing content distribution systems. We have designed and implemented a high-performance content distribution system, called V-Formation, based on the CPM. The CPM guides V-Formation toward a global allocation of bandwidth that maximizes the aggregate download bandwidth of consumers. Results from a PlanetLab deployment and extensive simulations show that V-Formation achieves high aggregate bandwidth and that the CPM enables hosts to converge quickly on a stable allocation of resources in a wide range of deployment scenarios.

This paper introduces a lightweight, scalable and accurate framework, called Meridian, for performing node selection based on network location. The framework consists of an overlay network structured around multi-resolution rings, query routing with direct measurements, and gossip protocols for dissemination. We show how this framework can be used to address three commonly encountered problems, namely, closest node discovery, central leader election, and locating nodes that satisfy target latency constraints in large-scale distributed systems without having to compute absolute coordinates. We show analytically that the framework is scalable with logarithmic convergence when Internet latencies are modeled as a growth-constrained metric, a low-dimensional Euclidean metric, or a metric of low doubling dimension. Large scale simulations, based on latency measurements from 6.25 million node-pairs as well as an implementation deployed on PlanetLab show that the framework is accurate and effective.