Papers
TAO: Facebook's Distributed Data Store for the Social Graph
Nathan Bronson, Zach Amsden, George Cabrera, Prasad Chakka, Peter Dimov, Hui Ding, Jack Ferris, Anthony Giardullo, Sachin Kulkarni, Harry Li, Mark Marchukov, Dmitri Petrov, Lovro Puzar, Yee Jiun Song, Venkat Venkataramani · USENIX Annual Technical Conference (ATC) 2013, 2013
The problem: serving the social graph from a lookaside cache
Before TAO, Facebook's web servers read and wrote the social graph directly in MySQL, using memcache as a look-aside cache: the client code (PHP) queried the cache, and on a miss went to the database and repopulated the cache. This design is great for iterating quickly, but the paper points out three fundamental failures when the data is a graph. First, edge lists: a key-value cache does not represent lists well, so any change or query forces fetching or reloading the entire list, and coordinating concurrent incremental updates is complicated. Second, distributed control logic: because it runs on clients that do not talk to each other, there are more failure modes and it is hard to avoid thundering herds. Third, expensive read-after-write consistency: MySQL uses asynchronous master/slave replication, so a write takes time to show up on the local replica, and the previous remedy (remote markers) required tracking keys known to be stale. The underlying motivation is that each page is assembled by pulling data from the graph and checking privacy for the viewer at read time, which produces an extremely read-dominated workload. TAO was built as a service that directly implements the objects-and-associations model, moving the control logic into the cache itself.
Data model and API: objects, associations, and time-ordered lists
TAO models the graph with objects (typed nodes) and associations (typed, directed edges). Each object has a 64-bit id that is unique across all types and a type (otype). An association is identified by the triple source (id1), association type (atype) and destination (id2), and at most one association of a given type exists between two objects. Both can carry data as key-value pairs, and each association has a 32-bit time field that is central to queries. Bidirectional edges are modeled as two associations, and TAO can keep them in sync if the type has an inverse type configured. The stated goal is not to support all graph queries, but to offer enough expressiveness for most applications with an efficient implementation. Queries revolve around association lists: all associations of a pair (id1, atype), in descending time order, taking advantage of locality by creation time (most data is old, but most queries want the most recent content). The operations include assoc_get, assoc_count, assoc_range (by position and limit) and assoc_time_range (by time range). There is a per-atype limit on the size of a query, typically 6,000; larger lists require multiple queries. A notable omission in the object API is compare-and-set, whose usefulness is reduced by the eventual consistency semantics.
Architecture: shards, cache layer and leader/follower hierarchy
TAO separates two cache layers from a storage layer. Persistent storage remains MySQL, and the data is divided into logical shards; each object id embeds the shard that hosts it, and an association lives in the shard of its id1, so any association query can be served by a single server. The cache layer implements the full API: the cache servers form a tier, and each request is routed to a single server per shard. The caches understand the semantics of what they store, so they can answer queries never seen before (for example, a cached count equal to zero is enough to answer an empty range). To avoid huge tiers, which suffer from hot spots and quadratic all-to-all communication, the cache is split into a leader level and several follower tiers. Clients talk only to followers; on a read miss or a write, the follower forwards to the leader, which accesses the database. Because all writes for a shard go through a single leader, it serializes concurrent writes and also protects the database from thundering herds, by limiting overlapping and pending queries. Writes to associations with an inverse may involve two shards, and TAO does not guarantee atomicity between the two updates: if there is a failure, an association without an inverse (hanging) may be left behind, repaired by an asynchronous job.
Geographic scale and consistency: eventual by default, strong on demand
With followers thousands of kilometers away from the databases, network round-trip time would become the bottleneck. Since the paper measures that read misses on followers are 25 times more frequent than writes, TAO adopts a master/slave architecture per shard: each region keeps a complete copy of the graph, reads and misses are served locally, and only writes are forwarded to the master region, so read latency is independent of the latency between regions. This favors performance and availability at the cost of potentially stale data. In normal operation, objects and associations are eventually consistent: after a write, TAO guarantees eventual delivery of invalidations or refills to all tiers, and replication lag is usually under one second. For the client that wrote, there is read-after-write consistency within a tier (with at most one failure per request), because the master leader returns a changeset that updates the cache synchronously. Invalidation and refill messages travel embedded in the database's replication stream, so as not to create inconsistency when reading from a replica that is still stale. For the small subset of reads that need more, TAO offers reads marked as critical, which are forwarded to the master region, where the consistent source of truth lives. The paper cites authentication as an example: there, the critical read is used so that replication lag does not allow stale credentials to be used.
Fault tolerance, hot spots and what the paper decides to sacrifice
TAO prioritizes availability and performance: when something fails, it prefers to keep rendering Facebook even with stale data. Its servers use aggressive network timeouts, keep per-destination timeouts, mark a host as down after several consecutive timeouts and remember it so future requests abort quickly, actively probing the failed machines to learn whether they are back. When failures are detected, the system routes around them on a best-effort basis, trading consistency for availability. Databases are marked as down in a global configuration if they fail, are taken out for maintenance, or fall too far behind in replication, and a shard's master region is switched automatically when a database fails. Writes that fail during this switch are reported to the client as failures and are not retried. On performance, the paper describes shard cloning, in which reads of a hot shard are served by several followers, and a small client-side cache for highly accessed objects, with a version number to omit unchanged data in responses. For high-degree objects (more than 6,000 associations of the same type), TAO does not cache the full list, and assoc_get queries that return empty always go to the database; the mitigation is to adjust the client code, for example by choosing the direction of the query or using the object's creation time to bound the search.
See the ideas at work
Practice the designs that use this paper in the System Design school.