Monarch is a globally-distributed in-memory time series database system in Google. Monarch runs as a multi-tenant service and is used mostly to monitor the availability, correctness, performance, load, and other aspects of billion-user-scale applications and systems at Google. Every second, the system ingests terabytes of time series data into memory and serves millions of queries. Monarch has a regionalized architecture for reliability and scalability, and global query and configuration planes that integrate the regions into a unified system. On top of its distributed architecture, Monarch has flexible configuration, an expressive relational data model, and powerful queries. This paper describes the structure of the system and the novel mechanisms that achieve a reliable and flexible unified system on a regionalized distributed architecture. We also share important lessons learned from a decade's experience of developing and running Monarch as a service in Google.

As distributed real-time applications gain in popularity, a key challenge is to allocate resources so that diverse real-time requirements (including non-real-time applications), distributed application components and varying workloads can all be accommodated without violating timeliness constraints. We examine the problem of resource allocation in distributed soft real-time systems, where both network and CPU resources are consumed. The timeliness constraints of applications are expressed through utility functions, which compute "benefit" as a function of end-to-end latency. We present LLA (Lagrangian Latency Assignment), a scalable and efficient distributed algorithm which maximizes aggregate utility by computing an optimal trade-off between end-to-end latency and allocated resources. The algorithm runs continuously and adapts to both workload and resource variations. LLA is guaranteed to converge if the workload and resource requirements stabilize. We evaluate the quality of results and convergence characteristics under various workloads, using both simulation and real-world experimentation.

Mashup applications mix and merge content (data and code) from multiple content providers in a user's browser, to provide high-value web applications that can rival the user experience provided by desktop applications. Current browser security models were not designed to support such applications and they are therefore implemented with insecure workarounds. In this paper, we present a secure component model, where components are provided by different trust domains, and can interact using a communication abstraction that allows ease of specification of a security policy. We have developed an implementation of this model that works currently in all major browsers, and addresses challenges of communication integrity and frame-phishing. An evaluation of the performance of our implementation shows that this approach is not just feasible but also practical.

Event-driven distributed infrastructures are becoming increasingly important for information dissemination and application integration. We examine the problem of optimal resource allocation for such an infrastructure composed of an overlay of nodes. Resources, like CPU and network bandwidth, are consumed by both message flows and message consumers; therefore, we consider both rate control for flows and admission control for consumers. This makes the optimization problem difficult because the objective function is nonconcave and the constraint set is nonconvex. We present LRGP (Lagrangian Rates, Greedy Populations), a scalable and efficient distributed algorithm to maximize the total system utility. The key insight of our solution involves partitioning the optimization problem into two types of subproblems: a greedy allocation for consumer admission control and a Lagrangian allocation to compute the flow rates, and linking the subproblems in a manner that allows tradeoffs between consumer admission and flow rates while satisfying the nonconvex constraints. LRGP allows an autonomic approach to system management where nodes collaboratively optimize aggregate system performance. We evaluate the quality of results and convergence characteristics under various workloads.

We describe algorithms to scalably support durable subscriptions in a publish-subscribe system. Durable subscriptions are guaranteed exactly-once message delivery, despite periods of disconnection from the system. Our algorithms persistently log each message only once in the system, and can support administratively specified 'early-release' policies that reclaim persistent storage in the presence of misbehaving durable subscribers. To efficiently recover messages missed by a disconnected durable subscriber, without refiltering messages published while the subscriber was disconnected, we persistently log filtering information in a manner optimized for the read/write pattern of durable subscriptions. Consolidation of data-structures across all subscribers that are done with catching up (after a disconnection) , helps the system support a larger number of subscribers. We experimentally demonstrate the low-latency and scalability properties of our implementation, both in the presence and absence of failures.

This paper presents a general knowledge model for propagating information in a content-based publish-subscribe system. The model is used to derive an efficient and scalable Protocol for exactly-once delivery to large numbers (tens of thousands per broker) of content-based subscribers in either publisher order or uniform total order Our protocol allows intermediate content filtering at each hop, but requires persistent storage only at the publishing site. It is tolerant of message drops, message reorderings, node failures, and link failures, and maintains only "soft" state at intermediate nodes. We evaluate the performance of our implementation both under failure-free conditions and with fault injection.

Interactive distributed applications are a relatively new class of applications that are enabled by sharing continuously evolving data across distributed sites (and users). The characteristics of application data include very fine-grained updates that can atomically access a subset of the shared data, masking of update effects, and irregular locality and contention for access. Existing programming approaches are not appropriate for programming such continuous shared data in a wide-area environment.

No abstract available.

The paper presents our state sharing support for building object oriented interactive groupware in wide area distributed environments. We motivate and present an asynchronous model for updating replicated state, which supports atomicity of updates across multiple shared objects. Coupled with our flexible marshalling framework, this model allows existing application data structure classes to be easily extended and made shareable. To solve the problem of replica consistency we use a novel combination of three mechanisms: (1) global locks, (2) detection of incorrect update ordering, and (3) cloning a subset of the shared objects and state reinitialization. Finally, to reduce network load due to fine grain user interaction, we provide a framework for application specified event batching called Late Event Modification.

No abstract available.