Papers
Large-scale Incremental Processing Using Distributed Transactions and Notifications
Daniel Peng, Frank Dabek · 9th USENIX Symposium on Operating Systems Design and Implementation (OSDI) 2010, 2010
The problem: updating a giant repository without reprocessing all of it
The paper starts from the task of maintaining the web search index while new pages are crawled. The indexing system must preserve invariants: if the same content appears at several URLs, only the one with the highest PageRank enters the index, and links are inverted so that anchor text is associated with the page being pointed to, even through duplicates. Before the work described, this was done by a series of MapReduce operations over the entire repository. In batch, the invariants are easy to maintain, because each stage finishes for all documents before the next one starts. The cost is that running the MapReduces only over the new pages is not enough, since there are links between them and the rest of the web; everything has to be reprocessed, which throws away the work of previous runs and makes latency proportional to the size of the repository, not the size of the update. Storing everything in a database with transactions doesn't work either, because the DBMSs of the time can't handle the volume: Google's indexing system stores tens of petabytes across thousands of machines. Distributed storage systems like Bigtable do scale, but they don't offer tools to maintain invariants under concurrent updates. Percolator fills this gap with two abstractions: ACID transactions over a random-access repository, and observers, pieces of code triggered when a column changes. The authors say it does not replace existing solutions for most tasks: what can't be split into small updates (sorting a file) is better served by MapReduce, and those who don't need strong consistency can use Bigtable alone.
Architecture: a layer on top of Bigtable and GFS
Percolator is a client library built on top of Bigtable. Three binaries run on each machine in the cluster: a Percolator worker, a Bigtable tablet server and a GFS chunkserver. Observers are linked into the worker, which scans Bigtable for changed columns (the notifications) and calls the corresponding observer as a function in the same process. Observers run transactions by sending reads and writes to the tablet servers, which in turn talk to the chunkservers. The system also depends on two small services: the timestamp oracle, which provides strictly increasing timestamps, and a lightweight lock service, used by workers to make the search for pending notifications more efficient. To the programmer, the repository is a handful of tables; each table is a collection of cells indexed by row and column, and each cell holds an uninterpreted array of bytes. Internally, to allow snapshot isolation, each cell is a series of values indexed by timestamp. Since Bigtable only offers atomic transactions within a single row, Percolator's challenge is to deliver what is missing: transactions across multiple rows and the observer framework. Two design decisions become clear. First, the system was designed for enormous scale without requiring extremely low latency, which allows it to lazily clean up locks left behind by transactions on failed machines, a delay that can postpone a commit by tens of seconds. Second, there is no central location for transaction management, nor a global deadlock detector: this increases the latency of conflicting transactions, but lets the system scale to thousands of machines.
Transactions with snapshot isolation and two-phase commit
Percolator offers cross-row and cross-table transactions with ACID snapshot-isolation semantics. Each transaction reads at a start timestamp, obtained from the oracle, and writes at a later commit timestamp. If two concurrent transactions write to the same cell, at most one commits. This is not serializability: snapshot isolation is subject to write skew. The advantage, the authors say, is that reads become more efficient, since any timestamp represents a consistent snapshot and reading a cell does not require acquiring locks. Because any node can modify Bigtable directly, there is no point at which to intercept traffic and grant locks, so Percolator keeps the locks explicitly, in special Bigtable columns, and they must survive machine failures. Each Percolator column becomes three Bigtable columns: data, lock, and write. Writes made with Set() are buffered until commit, which is a client-coordinated two-phase commit. In the prewrite phase, the transaction tries to lock all the cells it writes: it aborts if it sees a write record later than its start timestamp (a write conflict) or any lock, at any timestamp; otherwise it writes the data and the lock. In the second phase, it obtains the commit timestamp and, cell by cell, starting with the primary, replaces the lock with a write record that points to the timestamp where the data lives. A Get() first checks for a lock in the snapshot's visible range and, if there is one, waits for it to be released.
Client failures, primary lock and the timestamp oracle
If a client crashes in the middle of a commit, it leaves behind locks that would block future transactions. Percolator cleans this up lazily: a transaction A that encounters a conflicting lock left by B can conclude that B has failed and erase its locks. Since A can never be completely sure, it is necessary to avoid a race between A cleaning up and a B that has not really failed committing. The solution is to choose one cell of each transaction as a synchronization point, whose lock is the primary lock; the location of the primary is written into all the other locks. Both the commit and the cleanup must modify the primary inside a Bigtable row transaction, so only one of the two succeeds. If the client crashes after the commit point (at least one write record already exists), the lock is rolled forward: whoever encounters the lock inspects the primary and, if it has become a write record, the transaction committed. Otherwise, it rolls back, which is safe because the primary is always committed first. To estimate whether a worker has died, a token in Chubby is used, automatically deleted when the process exits, plus the wall time written into the lock, to catch workers that are alive but stalled. The oracle hands out timestamps in strictly increasing order and is contacted twice per transaction. It allocates ranges by writing the largest allocated value to stable storage, serves requests from memory and, after restarting, jumps forward, never backward. Each worker batches requests into a single outstanding RPC, and the oracle serves about 2 million timestamps per second on a single machine.
Notifications and observers: structuring incremental computation
Transactions maintain invariants, but the user also needs a way to trigger work. In Percolator, the programmer writes observers: each one registers a function and a set of columns, and the system invokes it after data is written to one of those columns in any row. An application is a chain of observers, each creating work for the next ones by writing to the table; in the indexing system, a MapReduce loads the documents through load transactions, which trigger the document processor, which triggers the clustering, and so on. Unlike database triggers, the notifications are not meant to maintain invariants: the observer runs in a transaction separate from the write that triggered it, so the two are not atomic. The guarantee given is that at most one observer transaction commits for each observed column change; several writes, however, can cause a single invocation, which is called message collapsing. To that end, each observed column has an acknowledgment (ack) column with the last start timestamp at which the observer ran. To find dirty cells among trillions, Percolator keeps a notify column, a plain Bigtable column with no transactional properties, which serves only as a hint for the scanner to check the ack column and decide whether the observer should run, in a separate locality group; workers scan it starting from random points, first taking a lock from the lock service so as not to duplicate work. There was also a tweak: the scanning threads clustered in a few regions (herd effect), and they now pick a new random position when they detect that they are scanning the same row as another thread.
Results and trade-offs of the indexer conversion
Percolator was built to create Google's base index, which was previously produced by a series of 100 MapReduces, in which each document spent 2 to 3 days being indexed. The system built on it, called Caffeine, crawls the same number of documents, but passes each one through Percolator as soon as it is crawled: the median is more than 100 times faster, and the paper's abstract says that the average age of documents in search results dropped by 50%. The price is resource consumption: Caffeine uses about twice as many resources for the same crawl rate, and the document collection was, at the time of the paper, 3 times larger than that of the previous system. In a synthetic benchmark with 240 machines and a repository of one billion documents, with 1% of the repository crawled per hour, MapReduce takes about 30 minutes on average to cluster a document, while Percolator takes about 2 seconds, roughly 1000 times less. At high crawl rates MapReduce becomes better, because reading data sequentially is cheaper per byte than doing random lookups; in that configuration Percolator saturated at 40% per hour. In a single-cell write microbenchmark, the overhead over plain Bigtable is about four times. In a benchmark based on TPC-E, performance scales almost linearly from 11 to 15,000 cores, reaching 11,200 transactions per second, but with an estimated cost of about 30 times more CPU per transaction than the reference commercial system, the cost of the layers.
See the ideas at work
Practice the designs that use this paper in the System Design school.