AI inference workflows are typically structured as a pipeline or graph of AI programs triggered by events. As events occur, the AIs perform inference or classification tasks under time pressure to respond or take some action. Standard techniques that reduce latency in other streaming settings (such as caching and optimization-driven scheduling) are of limited value because AI data access patterns (models, databases) change depending on the triggering event: a significant departure from traditional streaming. In this work, we propose a novel affinity grouping mechanism that makes it easier for developers to express application-specific data access correlations, enabling coordinated management of data objects in server clusters hosting streaming inference tasks. Our proposals are thus complementary to other approaches such as caching and scheduling. Experiments confirm the limitations of standard techniques, while showing that the proposed mechanism is able to maintain significantly lower latency as workload and scale-out increase, and yet requires only minor code changes.

Dairy owners invest heavily to keep their animals healthy. There is good reason to hope that technologies such as computer vision and artificial intelligence (AI) could reduce costs, yet obstacles arise when adapting these advanced tools to farming environments. In this work, we applied AI tools to dairy cow teat localization and teat shape classification, obtaining a model that achieves a mean average precision of 0.783. This digital twin-driven approach is intended as a first step towards automating and accelerating the detection and treatment of hyperkeratosis, mastitis, and other medical conditions that significantly burden the dairy industry.

Modern networking technologies such as Remote Direct Memory Access (RDMA) promise huge speedups in I/O bound platforms, but software layering overheads must first be overcome. Our paper studies this issue in a system that replicates small data objects using atomic multicast: a case in which internal synchronization is unavoidable, and any delay will be particularly impactful. Spindle, the methodology we propose, entails a series of optimizations including memory polling integrated with novel sender and receiver batching techniques, null-message send logic, and improved multi-thread synchronization. We applied Spindle to Derecho, an open-source library for atomic multicast, and obtained significant performance improvements both for the library itself and for an OMG-compliant avionics DDS layered on it. Derecho’s multicast bandwidth utilization for 10KB messages rose from 1GB/s to 9.7GB/s on a 12.5GB/s network, and it became more robust to delays even as latency dropped by nearly two orders of magnitude. While our focus is on the Derecho library and the OMG DDS, the same techniques should be relevant to databases, file systems, and IoT infrastructures.

Geo-replication is essential in reliable large-scale cloud applications. We argue that existing replication solutions are too rigid to support today’s diversity of data consistency and performance requirements. Stabilizer is a flexible geo-replication library, supporting user-defined consistency models. The library achieves high performance using control-plane / data-plane separation: control events do not disrupt data flow. Our API offers simple control-plane operators that allow an application to define its desired consistency model: a stability frontier predicate. We build a wide-area K/V store with Stabilizer, a Dropbox-like application, and a prototype pub/sub system to show its versatility and evaluate its performance. When compared with a Paxos-based consistency protocol in an emulated Amazon EC2 wide-area network, experiments show that for a scenario requiring a more accurate consistency model, Stabilizer achieves a 24.75% latency performance improvement. Compared to Apache Pulsar in a real WAN environment, Stabilizer’s dynamic reconfiguration mechanism improves the pub/sub system performance significantly according to our experiment results.

Applications with internal substructure are common in the cloud, where many systems are organized as independently logged and replicated subsystems that interact via flows of objects or some form of RPC. Restarting such an application is difficult: a restart algorithm needs to efficiently provision the subsystems by mapping them to nodes with needed data and compute resources, while simultaneously guaranteeing that replicas are in distinct failure domains. Additional failures can occur during recovery, hence the restart process must itself be a restartable procedure. In this paper we present an algorithm for efficiently restarting a service composed of sharded subsystems, each using a replicated state machine model, into a state that (1) has the same fault-tolerance guarantees as the running system, (2) satisfies resource constraints and has all needed data to restart into a consistent state, (3) makes safe decisions about which updates to preserve from the logged state, (4) ensures that the restarted state will be mutually consistent across all subsystems and shards, and (5) ensures that no committed updates will be lost. If restart is not currently possible, the algorithm will await additional resources, then retry.

Applications that aggregate and query data from distributed embedded devices are of interest in many settings, such as smart buildings and cities, the smart power grid, and mobile health applications. However, such devices also pose serious privacy concerns due to the personal nature of the data being collected. In this article, we present an algorithm for aggregating data in a distributed manner that keeps the data on the devices themselves, releasing only sums and other aggregates to centralized operators. We offer two privacy-preserving configurations of our solution, one limited to crash failures and supporting a basic kind of aggregation; the second supporting a wider range of queries and also tolerating Byzantine behavior by compromised nodes. The former is quite fast and scalable, the latter more robust against attack and capable of offering full differential privacy for an important class of queries, but it costs more and injects noise that makes the query results slightly inaccurate. Other configurations are also possible. At the core of our approach is a new kind of overlay network (a superimposed routing structure operated by the endpoint devices). This overlay is optimally robust and convergent, and our protocols use it both for aggregation and as a general-purpose infrastructure for peer-to-peer communications.

Multicast patterns are common in cloud computing and datacenter settings. Applications and infrastructure tools such as Spark frequently move large objects around, update files replicated to multiple nodes, or push new versions of programs to compute nodes. Some applications use replication directly, for example to increase fault-tolerance or achieve parallelism. Implementations of Paxos, block chains and other libraries often employ a hand-built reliable multicast as a primitive. Yet operating systems continue to be focused on point-to-point communication solutions such as TCP or RDMA, a hardware layer with TCP-like semantics that offers zero copy transfers, but lacks a reliable multi-destination transfer capability. Our system, RDMC (RDMA Multicast), offers reliable multicast functionality constructed from RDMA unicast. We discuss design choices, present a theoretical analysis of RDMC's robustness to delays and slow network links, and report on experiments that evaluate RDMC over Mellanox RDMA.

Cloud computing services often replicate data and may require ways to coordinate distributed actions. Here we present Derecho, a library for such tasks. The API provides interfaces for structuring applications into patterns of subgroups and shards, supports state machine replication within them, and includes mechanisms that assist in restart after failures. Running over 100Gbps RDMA, Derecho can send millions of events per second in each subgroup or shard and throughput peaks at 16GB/s, substantially outperforming prior solutions. Configured to run purely on TCP, Derecho is still substantially faster than comparable widely used, highly-tuned, standard tools. The key insight is that on modern hardware (including non-RDMA networks), data-intensive protocols should be built from non-blocking data-flow components.

The coming generation of Internet-of-Things (IoT) applications will process massive amounts of incoming data while supporting data mining and online learning. In cases with demanding real-time requirements, such systems behave as smart memories: a high-bandwidth service that captures sensor input, processes it using machine-learning tools, replicates and stores "interesting" data (discarding uninteresting content), updates knowledge models, and triggers urgently-needed responses.

Many applications perform real-time analysis on data streams. We argue that existing solutions are poorly matched to the need, and introduce our new Freeze-Frame File System. Freeze-Frame FS is able to accept streams of updates while satisfying "temporal reads" on demand. The system is fast and accurate: we keep all update history in a memory-mapped log, cache recently retrieved data for repeat reads, and use a hybrid of a real-time and a logical clock to respond to read requests in a manner that is both temporally precise and causally consistent. When RDMA hardware is available, the write and read throughput of a single client reaches 2.6GB/s for writes and 5GB/s for reads, close to the limit (about 6GB/s) on the RDMA hardware used in our experiments. Even without RDMA, Freeze Frame FS substantially outperforms existing options for our target settings.

Read-only caches are widely used in cloud infrastructures to reduce access latency and load on backend databases. Operators view coherent caches as impractical at genuinely large scale and many client-facing caches are updated asynchronously with best-effort pipelines. Existing solutions that support cache consistency are inapplicable to this scenario since they require a round trip to the database on every cache transaction. Existing incoherent cache technologies are oblivious to transactional data access, even if the backend database supports transactions. We propose T-Cache, a novel caching policy for read-only transactions in which inconsistency is tolerable (won't cause safety violations) but undesirable (has a cost). T-Cache improves cache consistency despite asynchronous and unreliable communication between the cache and the database. We define cache-serializability, a variant of serializability that is suitable for incoherent caches, and prove that with unbounded resources T-Cache implements this new specification. With limited resources, T-Cache allows the system manager to choose a trade-off between performance and consistency. Our evaluation shows that T-Cache detects many inconsistencies with only nominal overhead. We use synthetic workloads to demonstrate the efficacy of T-Cache when data accesses are clustered and its adaptive reaction to workload changes. With workloads based on the real-world topologies, T-Cache detects 43 -- 70% of the inconsistencies and increases the rate of consistent transactions by 33 -- 58%.

Operators of the nationwide power grid use proprietary data networks to monitor and manage their power distribution systems. These purpose-built, wide area communication networks connect a complex array of equipment ranging from PMUs and synchrophasers to SCADA systems. Collectively, these equipment form part of an intricate feedback system that ensures the stability of the power grid. In support of this mission, the operational requirements of these networks mandates high performance, reliability, and security. We designed Iron Stack, a system to address these concerns. By using cutting-edge software defined networking technology, Iron Stack is able to use multiple network paths to improve communications bandwidth and latency, provide seamless failure recovery, and ensure signals security. Additionally, Iron Stack is incrementally deployable and backward-compatible with existing switching infrastructure.

Modern Web services rely extensively upon a tier of in-memory caches to reduce request latencies and alleviate load on backend servers. Within a given cache, items are typically partitioned across cache servers via consistent hashing, with the goal of balancing the number of items maintained by each cache server. Effects of consistent hashing vary by associated hashing function and partitioning ratio. Most real-world workloads are also skewed, with some items significantly more popular than others. Inefficiency in addressing both issues can create an imbalance in cache-server loads.

In smart power grids it is possible to match supply and demand by applying control mechanisms that are based on fine-grained load prediction. A crucial component of every control mechanism is monitoring, that is, executing queries over the network of smart meters. However, smart meters can learn so much about our lives that if we are to use such methods, it becomes imperative to protect privacy. Recent proposals recommend restricting the provider to differentially private queries, however the practicality of such approaches has not been settled. Here, we tackle an important problem with such approaches: even if queries at different points in time over statistically independent data are implemented in a differentially private way, the parameters of the distribution of the query might still reveal sensitive personal information. Protecting these parameters is hard if we allow for continuous monitoring, a natural requirement in the smart grid. We propose novel differentially private mechanisms that solve this problem for sum queries. We evaluate our methods and assumptions using a theoretical analysis as well as publicly available measurement data and show that the extra noise needed to protect distribution parameters is small.

This experience report presents the results of an extensive performance evaluation conducted using four open-source implementations of Paxos deployed in Amazon's EC2. Paxos is a fundamental algorithm for building fault-tolerant services, at the core of state-machine replication. Implementations of Paxos are currently used in many prototypes and production systems in both academia and industry. Although all protocols surveyed in the paper implement Paxos, they are optimized in a number of different ways, resulting in very different behavior, as we show in the paper. We have considered a variety of configurations and failure-free and faulty executions. In addition to reporting our findings, we propose and assess additional optimizations to existing implementations.

Applications used to evaluate next-generation electrical power grids(``smart grids'') are anticipated to be compute and data-intensive. In this work, we parallelize and improve performance of one such application which was run sequentially prior to the use of our cloud-based configuration. We examine multiple cloud computing offerings, both commercial and academic, to evaluate their potential for improving the turnaround time for application results. Since the target application does not fit well into existing computational paradigms for the cloud, we employ parallel scripting tool, as a first step toward a broader program of adapting portable, scalable computational tools for use as enablers of the future smart grids. We use multiple clouds as a way to reassure potential users that the risk of cloud-vendor lock-in can be managed. This paper discusses our methods and results. Our experience sheds light on some of the issues facing computational scientists and engineers tasked with adapting new paradigms and infrastructures for existing engineering design problems.

Some network protocols tie application state to underlying TCP connections, leading to unacceptable service outages when an endpoint loses TCP state during fail-over or migration. For example, BGP ties forwarding tables to its control plane connections so that the failure of a BGP endpoint can lead to widespread routing disruption, even if it recovers all of its state but what was encapsulated by its TCP implementation. Although techniques exist for recovering TCP state transparently, they make assumptions that do not hold for applications such as BGP. We introduce application-driven TCP recovery, a technique that separates application recovery from TCP recovery. We evaluate our prototype, TCPR, and show that it outperforms existing BGP recovery techniques.

This paper examines the workload of Facebook's photo-serving stack and the effectiveness of the many layers of caching it employs. Facebook's image-management infrastructure is complex and geographically distributed. It includes browser caches on end-user systems, Edge Caches at ~20 PoPs, an Origin Cache, and for some kinds of images, additional caching via Akamai. The underlying image storage layer is widely distributed, and includes multiple data centers.

Energy accounts for a significant fraction of the operational costs of a data center, and data center operators are increasingly interested in moving toward low-power designs. Two distinct approaches have emerged toward achieving this end: the power-proportional approach focuses on reducing disk and server power consumption, while the green data center approach focuses on reducing power consumed by support-infrastructure like cooling equipment, power distribution units, and power backup equipment. We propose an integrated approach, which combines the benefits of both. Our solution enforces power-proportionality at the granularity of a rack or even an entire containerized data center; thus, we power down not only idle IT equipment, but also their associated support-infrastructure. We show that it is practical today to design data centers to power down idle racks or containers-and in fact, current online service trends strongly enable this model. Finally, we show that our approach combines the energy savings of power-proportional and green data center approaches, while performance remains unaffected.

No abstract available.

We present a power-lean storage system, where racks of servers, or even entire data center shipping containers, can be powered down to save energy. We show that racks and containers are more than the sum of their servers, and demonstrate the feasibility of designing a storage system that powers them up and down on demand further, we show that such a system would save an order of magnitude more energy than current disk-based power-proportional storage systems. Our simulation results using file system traces from the Internet Archive show over 44% energy savings, a 5x improvement over disk-based power management systems, without performance impact. We explore the tradeoffs in choosing the right unit to power off/on, and present an automated framework to compute the optimal power management unit for different scenarios.

The global network of data centers is emerging as an important distributed systems paradigm-commodity clusters running high-performance applications, connected by high-speed “lambda” networks across hundreds of milliseconds of network latency. Packet loss on long-haul networks can cripple applications and protocols: A loss rate as low as 0.1% is sufficient to reduce TCP/IP throughput by an order of magnitude on a 1-Gb/s link with 50-ms one-way latency. Maelstrom is an edge appliance that masks packet loss transparently and quickly from intercluster protocols, aggregating traffic for high-speed encoding and using a new forward error correction scheme to handle bursty loss.

High-bandwidth, semi-private optical lambda networks carry growing volumes of data on behalf of large data centers, both in cloud computing environments and for scientific, financial, defense, and other enterprises. This paper undertakes a careful examination of the end-to-end characteristics of an uncongested lambda network running at high speeds over long distances, identifying scenarios associated with loss, latency variations, and degraded throughput at attached end-hosts. We use identical fast commodity source and destination platforms, hence expect the destination to receive more or less what we send. We observe otherwise: degraded performance is common and easily provoked. In particular, the receiver loses packets even when the sender employs relatively low data rates. Data rates of future optical network components are projected to outpace clock speeds of commodity end-host processors, hence more and more end-to-end applications will confront the same issue we encounter. Our work thus poses a new challenge for those hoping to achieve dependable performance in higher-end networked settings.

IP Multicast (IPMC) in data centers becomes disruptive when the technology is used by a large number of groups, a capability desired by event notification systems. We trace the problem to root causes, and introduce Dr. Multicast (MCMD), a system that eliminates the issue by mapping IPMC operations to a combination of point-to-point unicast and traditional IPMC transmissions guaranteed to be safe. MCMD optimizes the use of IPMC addresses within a data center by merging similar multicast groups in a principled fashion, while simultaneously respecting hardware limits expressed through administrator-controlled policies. The system is fully transparent, making it backward-compatible with commodity hardware and software found in modern data centers. Experimental evaluation shows that MCMD allows a large number of IPMC groups to be used without disruption, restoring a powerful group communication primitive to its traditional role.

We design and implement a novel class of highly precise network instrumentation and apply this tool to perform the first exact packet-timing measurements of a wide-area network ever undertaken, capturing 10 Gigabit Ethernet packets in flight on optical fiber. Through principled design, we improve timing precision by two to six orders of magnitude over existing techniques. Our observations contest several common assumptions about behavior of wide-area networks and the relationship between their input and output traffic flows. Further, we identify and characterize emergent packet chains as a mechanism to explain previously observed anomalous packet loss on receiver endpoints of such networks.

No abstract available.

A dramatic scale-up of distributed computing platforms is underway. Internet routers can contain hundreds or thousands of line cards. Cloud computing platforms may contain tens or even hundreds of thousands of machines. What is gluing all of this together? Multicast to support data replication, event streams, and coordination. Yet yesterday’s multicast protocols are poorly matched to this new generation of uses; so much so that many cloud platforms refuse to deploy multicast as such, and have instead resorted to clumsy alternatives, mapping multicast to TCP or even web services method invocations. This talk will explore inadequacies of existing protocols, early progress towards better ones, and the longer term research agenda.

The most commonly deployed web service applications employ client-server communication patterns, with clients running remotely and services hosted in data centers. In this paper, we make the case for Service-Oriented Collaboration applications that combine service-hosted data with collaboration features implemented using peer-to-peer protocols. Collaboration features are awkward to support solely based on the existing web services technologies. Indirection through the data center introduces high latencies and limits scalability, and precludes collaboration between clients connected to one-another but lacking connectivity to the data center. Cornellpsilas Live Distributed Objects platform combines web services with direct peer-to-peer communication to eliminate these issues.

There is a growing need for a new type of WS-*/SOA standards that could facilitate hierarchical, object-oriented composition of client-side executable code. This is especially true for the sorts of client-side logic embedded in AJAX and rich Internet applications, virtual worlds and MMORPGs; code that deals with issuing requests to servers, processing their responses, rendering UI, interacting with users, and processing asynchronous events from other client nodes. The paper offers an analysis of client-side composition patterns, a brief explanation why they lack adequate support from the existing web technologies, and design guidelines for client-side component integration environments to follow. The proposed guidelines have been successfully implemented in a prototype system. Our analysis is thus strongly rooted in reality; it is based on real experiences with concrete application scenarios. The paper concludes by highlighting the key architectural aspects of our implementation with respect to the principles listed earlier.

In industries such as healthcare, there is a need to electronically share privacy-sensitive data across distinct organizations. We show how this can be done while allowing organizations to keep their legacy databases and maintain ownership of the data that they currently store. Without sending or mirroring data to any trusted, centralized entity, we demonstrate how queries can be answered in a distributed manner that preserves the privacy of the original data. This paper explains our distributed query execution engine, outlines how to bootstrap the system when only real world identifiers such as a name and date-of-birth are initially known, and offers details on the tradeoff between privacy and performance. We evaluate the scalability of this approach through simulation.

Gossip-based protocols are increasingly popular in large-scale distributed applications that disseminate updates to replicated or cached content. GO (gossip objects) is a pernode gossip platform that we developed in support of this class of protocols. In addition to making it easy to develop new gossip protocols and applications, GO allows nodes to join multiple gossip groups without losing the appealing fixed bandwidth guarantee of gossip protocols, and the platform optimizes rumor delivery latency in a principled manner. Our heuristic is based on the observations that multiple rumors can often be squeezed into a single IP packet, and that indirect routing of rumors can speed up delivery. We formalize these observations and develop a theoretical analysis of this heuristic. We have also implemented GO, and study the effectiveness of the heuristic by comparing it to the more standard random dissemination gossip strategy via simulation. We also evaluate GO on a trace from a popular distributed application.

This paper presents a novel object-oriented approach to modeling the semantics of distributed multi-party protocols such as leader election, distributed locks or reliable multicast, and a programming language that supports it. The approach extends our live distributed objects (LO) model with the new concept of a distributed flow (DF), a stream of events that flow concurrently at multiple locations. DFs correspond to local variables, private fields, and method parameters in Java-like languages; they're means by which one stores and communicates state. Protocol instances correspond to Java objects; they consume and output flows; their internal states are encapsulated as internal flows, and their internal logic is represented as operations on flows. Our language provides a new type of concern separation: the semantic structure of protocols is decoupled from implementation details such as construction and maintenance of overlays, trees, and other structures used for scalability. These can be generated by the compiler or at deployment time. This can be done differently in different parts of the network, to match the local environment.

Code-Partitioning Gossip (CPG) is a novel technique to facilitate implementation and analysis of gossip protocols. A gossip exchange is a pair-wise transaction between two nodes; a gossip system executes an endless sequence of exchanges between nodes chosen by a randomized procedure. Using CPG, the effects of a gossip exchange are succinctly defined by a single function that atomically updates a pair of node states based on their previous values. This function is automatically partitioned via program slicing into executable code for the roles of gossip-initiator and gossip-recipient, and networking code is added automatically. CPG may have concrete benefits for protocol analysis and authoring composite gossip protocols.

Peer-to-peer (P2P) architectures are popular for tasks such as collaborative download, VoIP telephony, and backup. To maximize performance in the face of widely variable storage capacities and bandwidths, such systems typically need to shift work from poor nodes to richer ones. Similar requirements are seen in today's large data centers, where machines may have widely variable configurations, loads, and performance. In this paper, we consider the slicing problem, which involves partitioning the participating nodes into k subsets using a one-dimensional attribute, and updating the partition as the set of nodes and their associated attributes change. The mechanism thus facilitates the development of adaptive systems. We begin by motivating this problem statement and reviewing prior work. Existing algorithms are shown to have problems with convergence, manifesting as inaccurate slice assignments, and to adapt slowly as conditions change. Our protocol, Sliver, has provably rapid convergence, is robust under stress and is simple to implement. We present both theoretical and experimental evaluations of the protocol.

Gossip-based communication protocols are attractive in cases where absolute delivery guarantees are not required due to their scalability, low overhead, and probabilistically high reliability. In earlier work, a gossip-based protocol known as gravitational gossip was created that allows the selection of quality ratings within subgroups based on workload and information update frequency. This paper presents an improved protocol that adds an adaptive component that matches the actual subgroup communication rates with desired rates coping with network variations by modifying underlying gossip weights. The protocol is designed for use in environments where many information streams are being generated and interest levels vary between nodes in the system. The gossip-based protocol is able to allow subscribers to reduce their expected workload in return for a reduced information rate. The protocol is a good fit for applications such as military information systems, sensor networks, and rescue operations. Experiments were conducted in order to compare the merits of different adaptation mechanisms. Experimental results show promise for this approach.

Soft state in the middle tier is key to enabling scalable and responsive three tier service architectures. While soft-state can be reconstructed upon failure, replicating it across multiple service instances is critical for rapid fail-over and high availability. Current techniques for storing and managing replicated soft state require mapping data structures to different abstractions such as database records, which can be difficult and introduce inefficiencies. Tempest is a system that provides programmers with data structures that look very similar to conventional Java Collections but are automatically replicated. We evaluate Tempest against alternatives such as in-memory databases and we show that Tempest does scale well in real world service architectures.

QSM is a multicast engine designed to support a style of distributed programming in which application objects are replicated among clients and updated via multicast. The model requires platforms that scale in dimensions previously unexplored; in particular, to large numbers of multicast groups. Prior systems werenpsilat optimized for such scenarios and canpsilat take advantage of regular group overlap patterns, a key feature of our application domain. Furthermore, little is known about performance and scalability of such systems in modern managed environments. We shed light on these issues and offer architectural insights based on our experience building QSM.

No abstract available.

Existing Web services notification and eventing standards are useful in many applications, but they have serious limitations precluding large-scale deployments: it is impossible to use IP multicast or for recipients to forward messages to others and scalable notification trees must be setup manually. We propose a design free of such limitations that could serve as a basis for extending or complementing these standards. The approach emerges from our prior work on QSM (Ostrowski et al., 2006), a new Web services eventing platform that can scale to extremely large environments

Future applications running on mobile platforms will sometimes need to query sensors and track sensor data over time. This paper uses the publish-subscribe paradigm as a natural solution to querying sensors from mobile platforms, and proposes a scalable approach to implement publish-subscribe, driven by the querying application. Our approach is evaluated by simulation, focusing on scalability

Flooding is an important communication primitive in mobile ad-hoc networks and also serves as a building block for more complex protocols such as routing protocols. In this paper, we propose a novel approach to flooding, which relies on proactive compensation packets periodically broadcast by every node. The compensation packets are constructed from dropped data packets, based on techniques borrowed from forward error correction. Since our approach does not rely on proactive neighbor discovery and network overlays it is resilient to mobilit.We evaluate the implementation of Mistral through simulation and compare its performance and overhead to purely probabilistic flooding. Our results show that Mistral achieves a significantly higher node coverage with comparable overhead.

Wireless networks present unusual challenges for mobile file system clients, since they are characterised by unpredictable connectivity and widely-varying bandwidth. The traditional approach to adapting network communication to these conditions is to write back file updates asynchronously when bandwidth is low. Unfortunately, this can lead to underutilisation of bandwidth and inconsistencies between clients. We describe a new mobile file system, MAFS, that supports graceful degradation of file system performance as bandwidth is reduced, as well as rapid propagation of essential file updates. MAFS is able to achieve 10-20% improvements in execution time for real-life file system traces featuring read-write contention.

Cognitive adaptive radio teams (CART) is a new platform developed by our group in support of collaborative mapping of complex communications-challenged environments, for example in support of search and rescue operations in environments lacking an adequate communications infrastructure. Experience during the 9/11 terrorist attacks, Asian tsunami, Kashmir earthquake, and post-Katrina Gulf Coast make it clear that rescue workers cannot count upon computer networks or even cell telephone support in the immediate aftermath of such events. Similarly, military urban warfare operations must also be conducted in locations lacking communication infrastructure. CART combines state-of-the-art ad-hoc networking technology with machine learning and prediction algorithms to offer new capabilities under these very difficult conditions

PLATO is a predictive total ordering protocol designed for low-latency multicast in datacenters. It predicts out-of-order arrival of multicast packets by observing their inter-arrival times, and delays packets before passing them up to the application only if it believes the packets to have arrived in the wrong order. We show through experimentation on real datacenter-style networks that the inter-arrival time of consecutive packet pairs is an excellent predictor of out-of-order delivery. We evaluate an implementation of PLATO on the Emulab testbed, and show that it drives down delivery latencies by more than a factor of 2 compared to the fixed-sequencer protocol

Data centers constructed as clusters of inexpensive machines have compelling cost-performance benefits, but developing services to run on them can be challenging. This paper reports on a new framework, the scalable services architecture (SSA), which helps developers develop scalable clustered applications. The work is focused on non-transactional high-performance applications; these are poorly supported in existing platforms. A primary goal was to keep the SSA as small and simple as possible. Key elements include a TCP-based "chain replication" mechanism and a gossip-based subsystem for managing configuration data and repairing inconsistencies after faults. Our experimental results confirm the effectiveness of the approach

Datacenters are complex environments consisting of thousands of failure-prone commodity components connected by fast, high capacity interconnects. The software running on such datacenters typically uses multicast communication patterns involving multiple senders. We examine the problem of time-critical multicast in such settings, and propose Slingshot, a protocol that uses receiver-based FEC to recover lost packets quickly. Slingshot offers probabilistic guarantees on timeliness by having receivers exchange FEC packets in an initial phase, and optional complete reliability on packets not recovered in this first phase. We evaluate an implementation of Slingshot against SRM, a well-known multicast protocol, and show that it achieves two orders of magnitude faster recovery in datacenter settings

Large-scale and dynamically changing distributed systems such as the Grid, peer-to-peer overlays, etc., need to collect several kinds of global statistics in a decentralized manner. In this paper, we tackle a specific statistic collection problem called Group Size Estimation, for estimating the number of non-faulty processes present in the global group at any given point of time. We present two new decentralized algorithms for estimation in dynamic groups, analyze the algorithms, and experimentally evaluate them using real-life traces. One scheme is active: it spreads a gossip into the overlay first, and then samples the receipt times of this gossip at different processes. The second scheme is passive: it measures the density of processes when their identifiers are hashed into a real interval. Both schemes have low latency, scalable perprocess overheads, and provide high levels of probabilistic accuracy for the estimate. They are implemented as part of a size estimation utility called PeerCounter that can be incorporated modularly into standard peer-to-peer overlays. We present experimental results from both the simulations and PeerCounter, running on a cluster of 33 Linux servers.

No abstract available.

With the widespread availability of wireless technology and the deployment of an increasing variety of sensors, information generated by sensors is becoming available to applications running on mobile nodes. Retrieving this information in a reliable, efficient manner will be an important building block for many applications. However, unreliable, low bandwidth communication links and node mobility make efficient, reliable, and scalable information retrieval a challenge.

The restructuring of the electric power grid has created new control and monitoring requirements for which classical technologies may be inadequate. The most obvious way of building such systems, using TCP connections to link monitoring systems with data sources, gives poor scalability and exhibits instability precisely when information is most urgently required. Astrolabe, Bimodal Multicast, and Gravitational Gossip, technologies of our own design, seek to overcome these problems using what are called "epidemic" communication protocols. This paper evaluates a hypothetical power monitoring scenario involving the New York State grid, and concludes that the technology is well matched to the need.

Rapid acceptance of the Web Services architecture promises to make it the most widely supported and popular object-oriented architecture to date. One consequence is that a wave of mission-critical Web Services applications will certainly be deployed in coming years. Yet the reliability options available within Web Services are limited in important ways. To use a term proposed by IBM, Web Services systems need to become far more autonomic, configuring themselves, diagnosing faults, and managing themselves. High availability applications need more attention. Moreover, the scenarios in which such issues arise often entail very large deployments, raising questions of scalability. In this paper we propose a path by which the architecture could be extended in these respects.

Applications on mobile computers must adapt to high variability in wireless network performance. Extending the semantics of transport protocols to offer more control over communication to the user allows applications to adapt their behavior to bandwidth variability. We examine adding bandwidth notifications, priorities and timeliness guarantees to a network API as a method for achieving greater application control over bursty traffic. Experiments demonstrate that the extended API allows applications to adjust to bandwidth variations effectively. We also compare three different implementations of the API: two which run on top of TCP, and one new protocol, ATP, which performs comparably to the TCP extensions, but has better performance for some workloads, including a workload simulating remote file system traffic.

Scalable management and self-organizational capabilities are emerging as central requirements for a generation of large-scale, highly dynamic, distributed applications. We have developed an entirely new distributed information management system called Astrolabe. Astrolabe collects large-scale system state, permitting rapid updates and providing on-the-fly attribute aggregation. This latter capability permits an application to locate a resource, and also offers a scalable way to track system state as it evolves over time. The combination of features makes it possible to solve a wide variety of management and self-configuration problems. This paper describes the design of the system with a focus upon its scalability. After describing the Astrolabe service, we present examples of the use of Astrolabe for locating resources, publish-subscribe, and distributed synchronization in large systems. Astrolabe is implemented using a peer-to-peer protocol, and uses a restricted form of mobile code based on the SQL query language for aggregation. This protocol gives rise to a novel consistency model. Astrolabe addresses several security considerations using a built-in PKI. The scalability of the system is evaluated using both simulation and experiments; these confirm that Astrolabe could scale to thousands and perhaps millions of nodes, with information propagation delays in the tens of seconds.

Reliable multicast delivery requires that a multicast message be received by all members in a group. Hence certain or all members need to buffer messages for possible retransmissions. Designing an efficient buffer management algorithm is challenging in large multicast groups where no member has complete group membership information and the delivery latency to different members could differ by orders of magnitude. We propose an innovative two-phase buffering algorithm, which explicitly addresses variations in delivery latency seen in large multicast groups. The algorithm effectively reduces buffer requirements by adaptively allocating buffer space to messages most needed in the system and by spreading the load of buffering among all members in the group. Simulation and experimental results demonstrate that the algorithm has good performance.

The paper discusses fault-tolerant, scalable solutions to the problem of accurately and scalably calculating global aggregate functions in large process groups communicating over unreliable networks. These groups could represent sensors or processes communicating over a network that is either fixed (e.g., the Internet) or dynamic (e.g., multihop ad-hoc). Group members are prone to failures. The ability to evaluate global aggregate properties (e.g., the average of sensor temperature readings) is important for higher-level coordination activities in such large groups. We first define the setting and problem, laying down metrics to evaluate different algorithms for the same. We discuss why the usual approaches to solve this problem are unviable and unscalable over an unreliable network prone to message delivery failures and crash failures. We then propose a technique to impose an abstract hierarchy on such large groups, describing how this hierarchy can be made to mirror the network topology. We discuss several alternatives to use this technique to solve the global aggregate function evaluation problem. Finally, we present a protocol based on gossiping that uses this hierarchical technique. We present mathematical analysis and performance results to validate the robustness, efficiency and accuracy of the Hierarchical Gossiping algorithm.

In recent years, a number of applications of ad-hoc networks have been proposed. Many of them are based on the availability of a robust and reliable multicast protocol. We address the issue of reliability and propose a scalable method to improve packet delivery of multicast routing protocols and decrease the variation in the number of packets received by different nodes. The proposed protocol works in two phases. In the first phase, any suitable protocol is used to multicast a message to the group, while in the second concurrent phase, the gossip protocol tries to recover lost messages. Our proposed gossip protocol is called Anonymous Gossip (AG) since nodes need not know the other group members for gossip to be successful. This is extremely desirable for mobile nodes, that have limited resources, and where the knowledge of group membership is difficult to obtain. As a first step, anonymous gossip is implemented over MAODV without much overhead and its performance is studied. Simulations show that the packet delivery of MAODV is significantly improved and the variation in number of packets delivered is decreased.

An efficient error recovery algorithm is essential for a liable multicast in large groups. Tree-based protocols (RMTP, TMTP, LBRRM) group receivers into local regions and select a repair server for performing error recovery in each region. Hence a single server bears the entire responsibility of error recovery for a region. In addition, the deployment of repair servers requires topological information of the underlying multicast tree, which is generally not available at the transport layer. This paper presents RRMP, a randomized reliable multicast protocol which improves the robustness of tree-based protocols by diffusing the responsibility of error recovery among all members in a group. The protocol works well within the existing IP multicast framework and does not require additional support from routers. Both analysis and simulation results show that the performance penalty due to randomization is low and can be tuned according to application requirements.

Ensemble is a Group Communication System built at Cornell and the Hebrew universities. It allows processes to create process groups within which scalable reliable fifo-ordered multicast and point-to-point communication are supported. The system also supports other communication properties, such as causal and total multicast ordering, flow control, and the like. This article describes the security protocols and infrastructure of Ensemble. Applications using Ensemble with the extensions described here benefit from strong security properties. Under the assumption that trusted processes will not be corrupted, all communication is secured from tampering by outsiders. Our work extends previous work performed in the Horus system (Ensemble's predecessor) by adding support for multiple partitions, efficient rekeying, and application-defined security policies. Unlike Horus, which used its own security infrastructure with nonstandard key distribution and timing services, Ensemble's security mechanism is based on off-the shelf authentication systems, such as PGP and Kerberos. We extend previous results on group rekeying, with a novel protocol that makes use of diamondlike data structures. Our Diamond protocol allows the removal of untrusted members within milliseconds. In this work we are considering configurations of hundreds of members, and further assume that member trust policies are symmetric and transitive. These assumptions dictate some of our design decisions.

An emerging generation of mission-critical networked applications is placing demands on the Internet protocol suite that go well beyond the properties they were designed to guarantee. Although the "next generation internet" (NGI) is intended to respond to the need, when we review such applications in light of the expected functionality of the NGI, it becomes apparent that the NGI will be faster but not more robust. We propose a new kind of virtual overlay network (VON) that overcomes this deficiency and can be constructed using only simple extensions of existing network technology. In this paper, we use the restructured electric power grid to illustrate the issues, and elaborate on the technical implications of our proposal.

This paper describes how experiences with building industrial strength distributed applications have dramatically changed the assumptions about what tools are needed to build these systems.

Process group toolkits provide methods to structure a system as a set of groups of cooperating processes, to detect process failures, and to order events (by ordering messages). Such tools have a performance cost for applications, particularly when a system is built using a large number of overlapping groups. We built an event-driven simulation to study performance of causally ordered message delivery in large systems composed of overlapping groups. Our studies, the first ever of multiple group systems, reveal some conditions under which the delays can be very large: two orders of magnitude greater than when delays are not required. Further, in a large system these delays can lead to increased system burstiness which limits system scalability. These results suggest that a system supporting multiple overlapping groups needs to be carefully designed and the system should often provide users with control over when to apply ordering guarantees.

Although building systems from components has attractions, this approach also has problems. Can we be sure that a certain configuration of components is correct? Can it perform as well as a monolithic system? Our paper answers these questions for the Ensemble communication architecture by showing how, with help of the Nuprl formal system, configurations may be checked against specifications, and how optimized code can be synthesized from these configurations. The performance results show that we can substantially reduce end-to-end latency in the already optimized Ensemble system. Finally, we discuss whether the techniques we used are general enough for systems other than communication systems.

By understanding how real users have employed reliable multicast in real distributed systems, we can develop insight concerning the degree to which this technology has matched expectations. This paper reviews a number of applications with that goal in mind. Our findings point to trade-offs between the form of reliability used by a system and its scalability and performance. We also find that to reach a broad user community (and a commercially interesting market) the technology must be better integrated with component and object-oriented systems architectures. Looking closely at these architectures, however, we identify some assumptions about failure handling which make reliable multicast difficult to exploit. Indeed, the major failures of reliable multicast are associated with attempts to position it within object-oriented systems in ways that focus on transparent recovery from server failures. The broader opportunity appears to involve relatively visible embeddings of these tools into object-oriented architectures enabling knowledgeable users to make trade-offs. Fault-tolerance through transparent server replication may be better viewed as an unachievable holy grail. Copyright © 1999 John Wiley & Sons, Ltd.

Maestro is a middleware support tool for distributed multimedia and collaborative computing applications. These applications share a common need for managing multiple subgroups while providing possibly different quality-of-service guarantees for each of these groups. Maestro's functionality maps well into these requirements, and can significantly shorten the development time of such applications. In this paper, we report on Maestro, and demonstrate its utility in implementing several multimedia and collaborative computing applications. In particular, we provide a detailed description of the implementation of IMUX, a pseudo X-server (proxy) for collaborative computing applications that is based on Maestro. Copyright © 1999 John Wiley & Sons, Ltd.

There are many methods for making a multicast protocol “reliable.” At one end of the spectrum, a reliable multicast protocol might offer tomicity guarantees, such as all-or-nothing delivery, delivery ordering, and perhaps additional properties such as virtually synchronous addressing. At the other are protocols that use local repair to overcome transient packet loss in the network, offering “best effort” reliability. Yet none of this prior work has treated stability of multicast delivery as a basic reliability property, such as might be needed in an internet radio, television, or conferencing application. This article looks at reliability with a new goal: development of a multicast protocol which is reliable in a sense that can be rigorously quantified and includes throughput stability guarantees. We characterize this new protocol as a “bimodal multicast” in reference to its reliability model, which corresponds to a family of bimodal probability distributions. Here, we introduce the protocol, provide a theoretical analysis of its behavior, review experimental results, and discuss some candidate applications. These confirm that bimodal multicast is reliable, scalable, and that the protocol provides remarkably stable delivery throughput.

Trends in networking and distributed computing are creating a new generation of applications that must adapt as the environment within which they execute changes. Examples of adaptation include switching protocols to overcome a security exposure or failure mode seen only in certain settings, changing data rates to accommodate a slow link, or adapting the behavior of a high level application to match the set of participants using the application. We describe the Ensemble system, a tool for building adaptive distributed programs. © 1998 John Wiley & Sons, Ltd.

The virtual synchrony model for group communication has proven to be a powerful paradigm for building distributed applications. Implementations of virtual synchrony usually require the use of failure detectors and failure recovery protocols. In applications that require the use of a large number of groups, significant performance gains can be attained if these groups share the resources required to provide virtual synchrony. A service that maps user groups onto instances of a virtually synchronous implementation is called a light-weight group service. This paper proposes a new design for the light-weight group protocols that enables the usage of this service in a transparent manner as a test case, the new design was implemented in the Horus system, although the underlying principles can be applied to other architectures as well. The paper also presents performance results from this implementation.

The Horus system supports a communication architecture that treats protocols as instances of an abstract data type.This approach encourages developers to partition complex protocols into simple microprotocols, each of which is implemented by a protocol layer.Protocol layers can be stacked on top of each other in a variety of ways, at run-time.First, we describe the classes of protocols that can be supported this way.Next, we present the Horus object model that we designed for this technology, and the interface between the layers that makes it all work.We then present an example layer that implements a group membership protocol.Next, we show how, given a set of required properties, an appropriate stack can be constructed.We look at an example stack of protocols, which provides fault-tolerant, totally ordered communication between a group of processes.The work contributes a standard framework for protocol development and experimentation, provides a high performance implementation of the virtual synchrony model, and introduces a methodology for increasing the robustness of the protocol development process.

Even as wireless networks create the potential for access to information from mobile platforms, they pose a problem for privacy. In order to retrieve messages, users must periodically poll the network. The information that the user must give to the network could potentially be used to track that user. However, the movements of the user can also be used to hide the user's location if the protocols for sending and retrieving messages are carefully designed. We have developed a replicated memory service which allows users to read from memory without revealing which memory locations they are reading. Unlike previous protocols, our protocol is efficient in its use of computation and bandwidth. We show how this protocol can be used in conjunction with existing privacy preserving protocols to allow a user of a mobile computer to maintain privacy despite active attacks.>

We devetop necessary conditions for the development of asynchronous distributed sofiware that will perform uniform actions (’evenis that if performed by any pro-cess, must be performed at all processes). The pa-per focuses on dynamic uniformity, which differs from ihe classical problems in that processes continually leave and join the ongoing computation. It relates the problem to asynchronous Consensus, and shows that Consensus is a harder problem. We provide a rigorous characterization of the framework upon which several existing distributed programming environments are based. And, our work shows that progress is some-times possible in a primary-partition model even when consensus is not. 1

How should distributed systems preserve consistency in the presence of concurrency and failures? For systems designed as assemblies of independently developed components, concurrent access to data or data structures would normally arise within individual programs, and be controlled using mutual exclusion constructs, such as semaphores and monitors. Where data is persistent and/or sets of operations are related to one another, transactions or linearizability may be more appropriate. Systems that incorporate cooperative styles of distributed execution often replicate or distribute data within groups of components. In these cases, group-oriented consistency properties must be maintained, and tools based on the virtual synchrony execution model greatly simplify the task confronting an application developer. All three styles of distributed computing are likely to be seen in future systems-often, within the same application. This leads us to propose an integrated approach that permits applications that use virtual synchrony to interact with concurrent objects that respect a linearizability constraint, and vice versa. Transactional subsystems are treated as a special case of linearizability.

Process groups are a common abstraction for fault-tolerant computing in distributed systems. We present a security architecture that extends the process group into a security abstraction. Integral parts of this architecture are services that securely and fault tolerantly support cryptographic key distribution. Using replication only when necessary, and introducing novel replication techniques when it was necessary, we have constructed these services both to be easily defensible against attack and to permit key distribution despite the transient unavailability of a substantial number of servers. We detail the design and implementation of these services and the secure process group abstraction they support. We also give preliminary performance figures for some common group operations.

We present a method for constructing replicated services that retain their availability and integrity despite several servers and clients being corrupted by an intruder, in addition to others failing benignly. We also address the issue of maintaining a causal order among client requests. We illustrate a security breach resulting from an intruder's ability to effect a violation of causality in the sequence of requests processed by the service and propose an approach to counter this attack. An important and novel feature of our techniques is that the client need not be able to identify or authenticate even a single server. Instead, the client is required to possess only a single public key for the service. We demonstrate the performance of our techniques with a service we have implemented using one of our protocols.

A distributed security architecture is proposed for incorporation into group oriented distributed systems, and in particular, into the Isis distributed programming toolkit. The primary goal of the architecture is to make common group-oriented abstractions robust in hostile settings in order to facilitate the construction of high-performance distributed applications that can tolerate both component failure and malicious attacks. These abstractions include process groups and causal group multicast. A delegation and access control scheme is also proposed for use in group-oriented systems. The focus is on the security architecture; particular cryptosystems and key exchange protocols are not emphasized.>

Agreement on the membership of a group of processes in a distributed system is a basic problem that arises in a wide range of applications. Such groups occur when a set of processes co-operate to perform some task, share memory, monitor one another, subdivide a computation, and so forth. In this paper we discuss the Group Membership Problem as it relates to failure detection in asynchronous, distributed systems. We present a rigorous, formal specification for group membership under this interpretation. We then present a solution for this problem that improves upon previous work.

The ISIS toolkit is a distributed programming environment based on support for virtually synchronous process groups and group communication. A suite of protocols is presented to support this model. The approach revolves around a multicast primitive, called CBCAST, which implements a fault-tolerant, causally ordered message delivery. This primitive can be used directly or extended into a totally ordered multicast primitive, called ABCAST. It normally delivers messages immediately upon reception, and imposes a space overhead proportional to the size of the groups to which the sender belongs, usually a small number. It is concluded that process groups and group communication can achieve performance and scaling comparable to that of a raw message transport layer. This finding contradicts the widespread concern that this style of distributed computing may be unacceptably costly.

We describe applications of a virtually synchronous environment for distributed programming, which underlies a collection of distributed programming tools in the ISIS2 system. A virtually synchronous environment allows processes to be structured into process groups, and makes events like broadcasts to the group as an entity, group membership changes, and even migration of an activity from one place to another appear to occur instantaneously — in other words, synchronously. A major advantage to this approach is that many aspects of a distributed application can be treated independently without compromising correctness. Moreover, user code that is designed as if the system were synchronous can often be executed concurrently. We argue that this approach to building distributed and fault-tolerant software is more straightforward, more flexible, and more likely to yield correct solutions than alternative approaches.

The design and correctness of a communication facility for a distributed computer system are reported on. The facility provides support for fault-tolerant process groups in the form of a family of reliable multicast protocols that can be used in both local- and wide-area networks. These protocols attain high levels of concurrency, while respecting application-specific delivery ordering constraints, and have varying cost and performance that depend on the degree of ordering desired. In particular, a protocol that enforces causal delivery orderings is introduced and shown to be a valuable alternative to conventional asynchronous communication protocols. The facility also ensures that the processes belonging to a fault-tolerant process group will observe consistent orderings of events affecting the group as a whole, including process failures, recoveries, migration, and dynamic changes to group properties like member rankings. A review of several uses for the protocols in the ISIS system, which supports fault-tolerant resilient objects and bulletin boards, illustrates the significant simplification of higher level algorithms made possible by our approach.

Many distributed systems replicate data for fault tolerance or availability. In such systems, a logical update on a data item results in a physical update on a number of copies. The synchronization and communication required to keep the copies of replicated data consistent introduce a delay when operations are performed. In this paper, we describe a technique that relaxes the usual degree of synchronization, permitting replicated data items to be updated concurrently with other operations, while at the same time ensuring that correctness is not violated. The additional concurrency thus obtained results in better response time when performing operations on replicated data. We also discuss how this technique performs in conjunction with a roll-back and a roll-forward failure recovery mechanism.

The ISIS system transforms abstract type specifications into fault-tolerant distributed implementations while insulating users from the mechanisms used tO achieve fault-toleram:e.This paper discusses tedmiques for obtaining a fault-tolerant implementation from a now distributed specification and for achieving improved performanc~ by concurrently updating replicated data.The system itself is based on a small set of communication primitives, which are interesting because they achieve high levels of concurrency while respecting higher level ordering requirements.The performance of distributed fault-tolerant services runnin 8 on this initial version of ISIS is found to be nearly as good as that of non-distributed, fault-intolerant ones.

This paper describes a technique for implementing k-resilient objects–distributed objects that remain available, and whose operations are guaranteed to progress to completion, despite up to k site failures. The implementation is derived from the object specification automatically, and does not require any information beyond what would be required for a nonresilient nondistributed implementation. It is therefore unnecessary for an applications programmer to have knowledge of the complex protocols nonnally employed to implement fault-tolerant objects. Our technique is used in ISIS, a system being developed at Cornell to support resilient objects.

Long-term electrocardiograms exhibit a small number of QRS morphologies (waveform shapes) whose analysis can reveal cardiac abnormalities. We considered the problem of accurately identifying instances of each in 24-h ECG recordings. A new learning algorithm was developed. Each QRS morphology is represented as a tree of rule activations, which associate attribute measurements with a rule. Each rule has a syntactic pattern together with a semantic procedure which manages and applies the knowledge stored in the activation. A single rule may be activated several times to learn different waveform segments. Delineation refinement improves each hypothesized signal interpretation. A simple conflict resolution mechanism resolves conflicting interpretations into a single unambiguous one. Comparison of the system with an existing program confirmed the promise of the new approach.

The design and implementation of a local network operating ystem based on the UNIX1operating system is described. UNIX has been extended to allow existing programs to access remote resources with no source program changes. Programs may access remote files, have a remote working directory, execute remote programs, and communicate with remote processes using the standard UNIX interprocess communication mechanism (pipe's). An efficient message-oriented interprocess communication mechanism and asynchronous I/O were added to the system to support the development of distributed applications and to make it easier to connect the local network to packet-switched networks.