A Tale of Two Production Bugs (Part 2)

Last time, I wrote about my experience debugging an issue where engineers commuting to and from the office caused production network blips. Ten years later, I encountered a bug at Amplitude that reminded me of that one and the fact that systems are often more intertwined than they seem at first glance.

At Amplitude, we used consul as our service discovery mechanism. We ran consul agents alongside our containers on EC2 instances, and those agents were responsible for reporting service status to the centralized consul servers. In particular, for our in-house columnar query system, we used service discovery with some statefulness (node ID) in order to distribute work in a cache-aware way. So each node, by nature of its ID, would consistently receive work that processed the same data files, and it would cache those files on its local SSDs to get good performance.

The bug occurred seemingly out of the blue, and the symptom was strange. A consul agent running on one of these query nodes would crash itself and then stop reporting that service's health. It would happen with varying frequency -- sometimes a few at once, sometimes days without issues. The system is designed for redundancy, so we could handle a handful of these, but losing a node would immediately start shifting load away and cause other nodes to begin processing and caching the data files. This would degrade performance in the system, and if left unchecked, sufficient node failures would cause queries to stop being processed completely.

Looking into the consul agent logs, the cause was that the consul network was reporting the same hostname with a different IP address than the actual node. For example, if query-12 was running at 10.1.2.3, the consul network was broadcasting that query-12 was actually at 10.1.2.4. When the agent running on query-12 received this message, essentially stating that it is not the consensus agent for that host, it would simply crash itself.

Naturally, we tried to look for the instance with IP address 10.1.2.4 to see why it was broadcasting the same hostname as the other instance. But it didn't exist. It turns out that 10.1.2.4 was the IP address of the previous instance with the same hostname before it had been rotated out (usually as part of regular security updates). That seemed promising, but that old instance had usually been gone for a long time already -- hours, if not days -- so it didn't make sense for it to start showing up again. Surely, we didn't have a rogue consul agent somehow still running on a terminated EC2 instance?

At this point, we simultaneously tried to investigate that possibility and dug through the consul source code to try to come up with hypotheses for how we could observe these symptoms. We happened to have upgraded consul in our infrastructure a few weeks before these errors started occurring, but the timing wasn't quite right. There was no evidence that the old instances were coming back according to any other metrics or logging that we had. Meanwhile, the frequency of the crashes was enough to be annoying, but not to cause real outages or customer impact. A few months passed without any plausible ideas for the root cause -- we kept restarting consul when it happened (painful!).

Finally, I got frustrated with this unexplained behavior and sat down one weekend to try and get to the bottom of it. After staring at hundreds of Datadog charts and all of the consul metrics for long enough, I found an unexplainable correlation. At the same time that these outdated IP broadcasts were happening, there were big spikes in the consul communication queues from the totally unrelated "inbound" cluster. The inbound cluster was an auto scaling group that grew and shrank with our incoming traffic. There was no obvious relationship between these communication queues and the agent crashes, but the correlation was so strong that I felt like I must be onto something.

As I dug into the inbound cluster, I stumbled across something curious. Some instances would be active, "disappear" from the metrics for a few hours or days, and then return again. That's not the typical behavior of an auto scaling group (terminated instances go away forever), and it was also suspiciously similar to the intervals that we were seeing the outdated IP broadcasts. Upon further examination, I found that we had enabled warm pools for the inbound auto scaling group, which is a feature that lets you keep a set of instances "warm" for faster auto scaling. In this particular case, warm meant that the instance was hibernated, i.e. its RAM contents saved for future use. When the instance came back, existing processes continued running as if they had never stopped. The warm pools were enabled right before we started observing the consul agent crashes (!).

At last, there was a root cause. The consul agents running in the inbound cluster were broadcasting outdated state after they came back from hibernation. The communication queues were growing because the consul agent briefly lost network connectivity and started buffering, only to empty those buffered messages (which could be very old) onto the network after the instance resumed. It's still not obvious to me why the consul network would allow these outdated messages to essentially overwrite known current state, but at that point I was ready to move past this issue. We initially disabled the warm pool, and eventually added a hook to shut down consul before hibernating, and the issue completely went away.

These two bugs really highlight the nature of debugging complex systems to me. If you have a bug that you can reproduce in a local environment, it's almost always straightforward to figure out. But in real production systems, the interactions are sometimes so wild that you could have never imagined the root cause at the beginning. In the decade between the last bug and this one, I became much more aware of this fact, and that open-mindedness helped me follow the seemingly nonsensical breadcrumbs that ultimately led to the answer. It's often said that debugging is significantly harder than writing code, and I'd say that my experiences corroborate that!

A Tale of Two Production Bugs (Part 1)

This is a recollection of a particularly tricky bug I encountered in a production system during my time at Sumo Logic in 2013, when I was just a few years out of school and still new to building software. Sumo Logic ran a very large AWS infrastructure in those days, spanning thousands of EC2 instances, as part of building the first log management system in the cloud. Being quite early in AWS' history (Sumo Logic was founded in 2010), there were a number of homegrown tools, including configuration management (a la Salt or Ansible) and a "VPN." Although I had used AWS previously, it was my first exposure to data processing and operations at scale.

I was about six months into the job when this bug started popping up. It was the worst type of symptom: seemingly random network connectivity drops across many different production services. Some services saw it more than others, and sometimes the drops would occur in bursts, but it happened pretty much every day. It was just innocuous enough (retries usually hid the problem) that we let it persist for about a month, with a few engineers taking a stab at debugging it here and there to no avail. The issue slowly became worse over time, and finally the team decided that we needed to track it down once and for all. I don't recall why (maybe nobody else wanted it?), but somehow the task got assigned to me.

Given how little I knew about networking and AWS at the time, it was quite a stretch for me. I spent a whole week playing with two EC2 instances and testing what could cause network disruptions between them. Eventually, I came across a plausible candidate: security group changes occasionally caused my test connection to drop in a non-deterministic way that resembled the bug we were seeing. But there was no obvious reason why the security groups would be changing in production so frequently -- after all, security groups are typically assigned when an instance starts up.

With this hypothesis in mind, I started to look more carefully at the production logs of the network errors. The distribution across services was mostly random, with the only correlation being the services with the most traffic showing up the most often (it still took me a while to rule service-specific reasons out). The distribution across time, however, was unusual. The errors occurred most frequently in two bursts during the day: one in the morning (8-9am), and one in the early evening (5-6pm). This didn't correlate with our actual traffic. And there were many fewer errors over the weekend, although traffic was noticeably lower in that case. The obvious correlation that these time windows brought to mind was... people going to and leaving work?

At this point, I felt like I was so close to figuring it out, but I was also at a loss for how to even find a next hypothesis to test. So I sent a message out to our internal chat (Campfire) asking if anyone could hazard a guess as to why engineers going to and leaving work would cause security group changes. Frankly, I felt a little silly asking such a strange question. To my surprise, our chief architect immediately responded, saying that he now knew the root cause of the bug.

It turned out that the homegrown "VPN" that I mentioned above wasn't a traditional VPN at all. Instead, the client running on each of our laptops would observe changes to our IP address and notify the server. The server would then modify a security group attached to all instances, which contained a list of developers' IP addresses, allowing production access. In fact, we had recently exceeded 50 engineers, and there was a limit of 50 rules per security group, meaning a second security group had been added, worsening the problem. So when each of us commuted to and from the office every day (as everyone did back then), we would unwittingly cause a small blip in the production network!

Everything fell into place after that, and we quickly switched to a real VPN. I still look back on this bug as a great learning experience as well as a satisfying "project," despite its nature being very different from writing code. It serves as a reminder to me to be open-minded when debugging, and to follow the patterns in the data, wherever they may lead. And little did I know that this episode would be the precursor to another one a decade down the road -- this time, with me as the chief architect seeing it all the way through.

Distributed Game Servers in KFChess

In the original implementation of kfchess.com, game state lived only in the memory of the single python process backend that ran the whole site. That keeps things very simple, but has the obvious drawback of having no good answers when the load on the site exceeds the capacity of that process (which it did on a small number of occasions). It also meant that any live games were lost when I deployed the site. In my recent rewrite, I wanted to solve both of these problems together by (a) enabling multiple game servers, (b) persisting game state across restarts, and (c) allowing game servers to act as redundancy for each other. This is a pretty standard distributed systems problem, but it's fun to see exactly how it plays out. I worked with Claude to design and implement it all, which you can read more about in this post.

Game servers, game states, and corresponding Redis keys.

To start off, we'll use Redis as the backend for the persistent state (AOF + RDB enabled). We store three types of keys: server heartbeats, game routing, and game snapshots. The server heartbeats are used to determine whether a server is alive, the game routing keys are used to know where to route requests to, and the game snapshots are used to restore the game state when it's lost. Note that in the design, games "belong" to a server at a particular point in time, and that server is responsible for all logic related to the game: advancing game ticks, handling requests, and running the AI.

That brings us to the problem: how do clients get routed to the correct game server? KFChess does in-game client-server communication via WebSockets, so once we establish a connection to the correct server, we're good. We use a caddy web server in front of all of the game servers, and normally it round-robins requests to the game servers. If a client tries to connect to a game and hits the correct server (has the game state in memory), then we're done. When the server doesn't have the game, it checks Redis for which server currently owns that game. If it finds the entry, then it sends a redirect back, which caddy handles for the client so that it connects to the correct server.

Game server redirect by looking up in Redis and routing to the right place.

This is omitted from the diagram to keep the complexity reasonable, but the initial server that handles the request also does a sanity check on the redirected server's heartbeat to make sure it's alive. If not, the initial server will try to claim the game as its own using atomic compare-and-set on the game routing entry in Redis, which is how games can move from one server to another. After successfully claiming a game, the server will load the snapshot state and continue the game from there -- in practice, we snapshot games every second, so it's possible to lose up to a second of game progress when this happens (can look like pieces "snapping" backwards). If the server fails to claim the game, it simply checks for redirect again.

One last point worth mentioning is that deploys are handled more cleanly than a true server-loss case. When a server goes through a clean shutdown, it snapshots all of its games into Redis, so the snapshots represent the latest game state. And in practice, the heartbeat threshold is high enough (5 seconds) so that the server isn't marked as dead on deploy. When the server comes back, it simply considers itself the owner of the same set of games, reloads the latest snapshots from Redis, and continues along. So no game state is lost at all, but there is a brief pause in the progression of the game as the server restarts.

That's the entire architecture for handling distributed game servers. It's relatively simple and incurs minimal overhead on request latency and CPU usage. I was pleasantly surprised by how nicely it ended up working, and it makes me a lot less hesitant to deploy code now!

Sparse File LRU Cache

An interesting file system feature that I came across a few years ago is sparse files. In short, many file systems allow you to create a logical file with "empty" (fully zeroed) blocks that are not physically backed until they get written to. Partially copying from ArchWiki, the behavior looks like this:

The file starts at 0 physical bytes on disk despite being logically 512MB. Then, after writing some non-zero bytes at a 16MB offset, it physically allocates a single block (4KB). The file system is maintaining metadata on which blocks of the file are physically represented on disk and which ones are not. To normal readers of the file, it's transparent -- the sparsity is managed completely by the file system.

At Amplitude, we found a cool use case for sparse files. All of the data is stored durably in cold storage (Amazon S3) in a columnar data format used for analytics queries. But it's inefficient and costly to fetch it from S3 every time, so the data gets cached on local NVMe SSDs (e.g. from the r7gd instance class). These local SSDs are more than ten times as expensive as cold storage, though, so you need a good strategy for deciding how and what to cache. To understand why sparse files are a good option for this, let's revisit columnar data formats briefly.

One of the observations about analytics queries that makes columnar data formats so effective is the fact that usually only a very small subset of columns (5-10 out of potentially thousands) are used on any particular query (and, to some extent, even across many queries). The data being stored in a columnar fashion means that each of these columns is a contiguous range inside the file, making it much faster to read. The contiguous ranges are also important in the context of our local caching -- we're not trying to pick out small pieces scattered across the file.

Setting sparse files aside for a moment, we originally had two different approaches for doing this caching:

• The most naive strategy is caching entire files from S3. This is simple and requires minimal metadata to manage, but has the obvious downside of wasting a lot of disk space on the SSDs by storing the columns that are rarely or never used.
• Another option is caching different columns as individual files on disk. This somewhat solves the wasted disk space issue, but now explodes the number of files, which requires a substantial amount of file system metadata. It also struggles with small columns, which are rounded up to the file system block size. With hundreds of thousands of customers of varying sizes, it's inevitable that the vast majority of files/columns are small.

At this point, it's pretty clear how sparse files give you an option in between these two. We imagine that we're caching entire files, except that the files are sparse, where only the columns that are used (more specifically, logical blocks that contain those columns) are physically present. This is simpler for a consumer to read and utilizes the disk better -- less file system metadata, and small columns are consolidated into file system blocks (as a bonus, this reduces S3 GETs as well). Similar to the latter approach above, consumers must declare the columns they need to the cache before reading them.

Managing sparse file block metadata in RocksDB.

This system requires us to manage metadata on which columns are cached, which we used a local RocksDB instance for. More specifically, we track metadata on logical blocks of these sparse files: which ones are present locally, and when they were last read. Using this, we approximate an LRU policy for invalidating data when the disk fills up (via periodic sweeps across all blocks). The invalidation process uses the fallocate system call with the FALLOC_FL_PUNCH_HOLE flag to signal to the file system that we want to reclaim that space.

As an important implementation detail, the logical blocks are variable-sized with a few smaller blocks at the head (still larger than file system blocks) and then larger blocks throughout the rest, which takes advantage of the fact that the file format has a metadata header (similar to Parquet) that always needs to be read to know how the columns are laid out. The variable-sized blocks are particularly suitable for the mix of very small and very large files that are present in the data.

This sparse file LRU cache improved many aspects of the query system simultaneously: fewer S3 GETs, less file system metadata, less file system block overhead, and fewer IOPS to manage the cache. In turn, that leads to significantly improved query performance at a lower cost. It's rare that a feature that lives as low-level as the file system has such a prominent impact on system design, so when it happens, it's pretty neat.

Distributed Systems and the CAP Theorem

In the field of distributed systems, the CAP theorem is an important result that often guides the design of such systems. The theorem states that a distributed system cannot satisfy consistency, availability, and partition tolerance simultaneously (see the Wikipedia article for definitions of these three properties). In practice, this typically means that, in the case of a network partition, there is some trade-off between consistency and availability. For example, HBase chooses consistency in that case, while Cassandra chooses availability (with eventual consistency). These days, many services have extremely high availability requirements, leading to the popularity of systems that sacrifice strong consistency for availability; having worked with Cassandra myself, however, it is clear that eventual consistency introduces a layer of complexity on both the client and server side that can be difficult to overcome. I was recently introduced to a blog post on how to beat the CAP theorem (it's lengthy, but well worth the read) written by Nathan Marz, the creator of Storm. He outlines an approach to building distributed systems that is intended to simplify the nature of eventual consistency.

First, let me preface the discussion. The article was met with some criticism, but from what I can tell it is mostly people not understanding his ideas combined with the provocative and ambiguous title. There's no such thing as "beating" the CAP theorem in the sense of violating it through some ingenious design; it's a theorem because someone proved it is always true and cannot be violated. The point being made is that we can address the CAP theorem in a way that doesn't lead us down a road of unmanageable complexity and, consequently, systems that are not robust and reliable.

The basic principle that the design leverages is that "data is inherently immutable." This is because data is always associated with a point in time; you can think of data as a fact that was true at that time. So while the balance of your bank account might change from $10$ to $20$ from time $T$ to time $T+1$, the two pieces of data, namely that your balance was $10$ at time $T$ and $20$ at time $T+1$, are forever true. In my experience, starting with this kind of definition sets you up for success because immutability is good, and it simplifies everything. From here, a distributed system is merely an accumulation of data that exposes methods of querying the data, where a query can be any arbitrary computation over all the data. The flexibility you have in querying the data is determined by what the system has chosen to expose, ranging from limited queries (e.g. a plain key-value store) to very expressive queries (e.g. SQL).

Now that we've gotten the mental model of distributed systems out of the way, let's take a look at the key piece of the design that let us "beat" the CAP theorem. Instead of treating all data homogeneously as most systems do, separate the data into two layers: the batch layer, say everything up until the last hour, and the real-time layer, i.e. everything from the last hour. Queries are then sent to both layers of the data and subsequently merged to produce the final result. Since queries are typically too slow to run across the entire batch layer all the time, we precompute "views" of the batch layer that allow queries to be quickly answered. For example, in a key-value store, the data is the history of all inserts, updates, and deletes, but we can precompute a view which is the current map of keys to values, which lets us answer any query by key quickly. Do this every hour, flushing the real-time layer as the view becomes available to query, and we have a system that only transiently depends on the real-time layer.

So what does this buy us? First of all, the batch layer is very simple. You have all of the data, you compute something from it, and you never have to deal with concurrency or anything of that sort. But we still need the real-time layer, which is going to be complex, so have we really saved anything? Let's think about failure (and if you're working with distributed systems, you should always be thinking about failure), both in terms of systems and humans. Failures will occur, and given the complexity of these real-time systems and the application code interacting with them, it's not unusual to corrupt or lose data in unexpected ways. The batch layer is essentially a reliable fallback mechanism in case of a catastrophe (which, as Marz recalls in an incident, can be as simple as running out of disk space somewhere). By isolating the complex real-time layer from the batch layer that is the ultimate source of truth, you protect yourself against these failures.

We can summarize this entire design exercise by the simple principle that we started out with: data is immutable (and immutability is good). Whether you're programming in a multithreaded environment or building a distributed data processing system, leverage immutability as much as you can. It simplifies things and protects you against your own mistakes. And again, I highly recommend checking out the full post to understand more of his thinking.

Bitcoin

The virtual currency bitcoin has been getting a lot of hype and news recently, as the price per coin has gone from about \$14 at the beginning of the year to a high of over \$1200 last week with a subsequent drop to its current value, which is between \$700 and \$800 (see here). While nobody really knows the future of bitcoin, the success of which depends highly on regulations, the technical aspects of its implementation are quite interesting. For the basic overview, first check out this explanation from the website, which the rest of this post assumes at least cursory knowledge of. The original bitcoin paper that contains all of the technical details I will be explaining can be found here. The core concept is a virtual currency that does not require a centralized authority to manage, which naturally makes it much more flexible and reduces the overhead of processing transactions. In order for such a system to work, there must be some source of trust, which is accomplished through a combination of public-key cryptography and a proof-of-work system.

Bitcoin, at its core, is really just a sequence of transactions between wallets, each of which is signed by the sender. Each wallet has a public key and a private key, and senders sign transactions with their private key such that everyone else can validate the transactions using the public key, as is typical with public-key cryptography. The problem that arises is that there is no easy way to verify that the sender did not "double-spend" their bitcoins, i.e. signing two transactions which send the same bitcoins. As it turns out, the only way to prevent such a situation without a centralized authority is to have a public, linear history of all transactions (known as the block chain). Assuming such a history, every new transaction can be verified to not include double-spending, and we have a working system.

So the remaining problem is how a distributed, peer-to-peer system can generate a transaction history that everyone agrees on. The solution is a proof-of-work system that is based on cryptographic hashing. At any point in time, bitcoin miners are continuously brute-forcing through hash computations in order to produce the next block in the chain. A block consists of the previous block hash, a set of transactions, and a nonce; it is only valid if its hash has sufficient difficulty, where difficulty is the number of 0's the hash begins with. Miners continually increment the nonce until they produce a block whose hash difficulty is accepted by the system (i.e. everyone else). The difficulty is chosen such that blocks are produced around every 10 minutes, which is why bitcoin transactions typically take that long to verify. Once a block has been committed to the chain, all the miners will generally accept it and move on to the next block, where there are some details about blocks being produced simultaneously, etc. The transactions that are part of the chain form the history from which future transactions can be verified. An example of a recently-produced block can be seen here.

There are a couple of remaining points that should be addressed when talking about bitcoin. Firstly, what is the incentive for people to mine (and thus produce the transaction history necessary for the system to function)? Bitcoin mining is the only way to produce new bitcoins; each block has a special transaction that gives the producer of the block some bitcoins, currently 25, although this reward decreases over time since the total number of bitcoins in circulation is capped at 21 million. The other incentive is transaction fees that are collected as payment for keeping the system running. Another issue is that the transaction history seems like it will grow infinitely and become unmanageable. This can be alleviated using the fact that transactions which are used as inputs to other transactions can be garbage-collected using Merkle trees (details in the paper). Lastly, it is important to consider the possibility of attackers producing fake blocks that reverse old transactions. The key point is that reversing an old transaction involves recomputing all of the blocks from that transaction onward since each block is a function of the previous one's hash. As such, the attacker would have to produce transaction history as quickly as the rest of the network combined, which seems to be a strong enough guarantee for people to trust bitcoin.

Whether bitcoin is a fad or not, it has a neat theoretical foundation, and I'm impressed by all of the work people are putting into it (blockchain.info is super cool). There is certainly room for disruption in the currency and payments industry, and it's only a matter of time before money is a purely virtual concept. Bitcoin also shows us that cryptography is a powerful tool, and it's worth thinking about how we can rely on mathematics for trust instead of other people (as depressing as that might sound).

Reactive Programming

There's an interesting idea being pushed forward by the Scala community about the way systems should be built, and it's called reactive programming. The reactive manifesto gives a nice summary of their views on the future of distributed applications and how the way we programmed ten years ago is inadequate in this future. Because of the changing landscape of latency expectations, data scale, and hardware improvements, going forward, systems must be event-driven, scalable, resilient, and responsive in order to meet user needs. There is a significant amount of ongoing Scala development in order to build tools that allow programmers to achieve these properties easily. This is the goal of the akka and spray projects.

The reactive programming model being adopted is based on actors. Given that I am a huge fan of the actor model for how it deals with the problems of concurrency, it is great to see an entirely new ecosystem developing around actors. The Scala community has really invested in actors as the new way in which communication between systems (or even within them) will be done in the future. From some slides for a recent talk on akka and spray, actors are part of a different programming paradigm that is "well-suited for building reactive systems" because they inherently shun the root of all concurrency evil, shared mutable state. Specifically, using actors versus the old model of managing threads in servlets gets around the high overhead of such management as well as the difficulties of synchronizing state. In some ways, we can view this as the next step in lightweight request handling: it was once (and still sometimes is) the case that each HTTP request would spawn a process, but that was greatly improved upon by having many threads within a single process handle requests. With actors, a single thread can handle many requests at the same time, allowing the resources to be much more efficiently utilized (this is the premise behind Node.js as well).

Spray is what you would expect the HTTP stack to look like on top of akka. While I haven't played with it myself, the code examples show that requests are mapped directly to actor messages, so everything is under the unified model of message handling. Judging from this, they have done a pretty good job from a performance standpoint as well, indicating just how lightweight request handling is in spray. Akka and spray together form a solid foundation on which you can build an application that conforms to the principles behind reactive programming. It is exciting to see technology evolving to meet the needs of users, programmers, and applications, and I'm sure that in a few years we'll see how far this paradigm can take us.

Gumball Technique

Using memcached on top of a traditional relational database (RDBMS) has become the most popular way of building large-scale web services, as it is both simple and fast in the majority of cases. Loads are typically read-heavy because for each piece of content produced, many people will consume it. Moreover, the most popular pieces of content will account for the majority of traffic. As such, having an in-memory caching layer that is able to efficiently serve the mass of read requests to the same data will naturally become standard. Cache consistency, however, is known to be a hard problem; in this case it involves synchronizing the state of two data stores that do not provide any transactional guarantees between them. Consider the following code for a client that talks to both memcached and an RDBMS.


Assuming that the cache and the RDBMS expose the same interface as the client itself, this code for handling puts and gets is straightforward. On a get, check if the cache has the value; if not, go to the database and cache the value retrieved if it exists. On a put, issue a put on the database and invalidate the entry in the cache. In barely 10 lines of real code, we already see a race condition that can result in stale cache data. For example, suppose the database currently has the pair ("k", "v1"), the cache does not have "k", and two clients A and B are simultaneously getting the key "k" and putting the pair ("k", "v2"), respectively. Then the following sequence is possible:

1. A sees that the cache does not contain the key "k".
2. A gets the value "v1" from the RDBMS.
3. B puts the pair ("k", "v2") into the RDBMS.
4. B removes the key "k" from the cache (was not there in the first place).
5. A caches the pair ("k", "v1") and returns.

So client A receives the value "v1", which is fine, but it also leaves that value in the cache after B has already updated it, meaning that future gets for key "k" will return stale data. It turns out that using memcached on top of an RDBMS is not quite as trivial as the code snippet above; to fix this race condition we can leverage the gumball technique.

The idea behind the gumball technique is clever: the cache server itself sees all of these operations (the get that misses, the remove, and then the put from the original getter), so it can deduce which ones will result in stale data. The only change to the protocol itself is to return a miss time $T_m$ (server time) on the first get and require the client to pass $T_m$ with the corresponding put, which we store along with the value in the cache. The rest of the work is done by the cache server itself and consists of two parts: managing "gumball" entries and rejecting certain put operations. Gumball entries are timestamped removal markers; every time a client issues a remove on a key, instead of just deleting the key, the server puts one of these gumballs that has the timestamp of the removal (again, server time). The time-to-live (TTL) of the entry, which we will refer to as $\Delta$, has a special way of being determined, but we can assume it is just some value that exceeds the typical length of a RDBMS query.

The tricky part of the algorithm comes when a put operation is issued (with a corresponding $T_m$ value). Let $T_C$ be the current time on the cache server. If $T_C - T_m > \Delta$, we ignore the put operation because a potential gumball entry could have been created in the window that has already expired. This is why $\Delta$ should be larger than the typical get/put cycle. If there is a gumball entry whose timestamp exceeds $T_m$, then it is likely that the race condition occurred and we should ignore the put operation. Lastly, if there is an actual value with miss time greater than $T_m$, we also have to ignore the put operation as the other put may have overridden a gumball entry that would have invalidated the current put. If none of these conditions hold, then we insert the key/value pair into the cache as normal (along with $T_m$).

These steps guarantee that the cache will never have a stale entry that persists following an update or delete, which brings back the consistency provided by an RDBMS in the first place. The paper goes into a bit more detail about how to choose $\Delta$ properly via a sliding window algorithm with some buffer, but the exact value is not important for correctness. It also discusses the performance impact, which seems to be negligible under normal loads, but does depend on how long the RDBMS queries take relative to the frequency of gets. The gumball technique is a nice example of leveraging minimal centralization and carefully observing access patterns to produce a simple way of ensuring distributed consistency. It would be great if memcached implementations actually started doing this, but since it requires an API change, it will probably not become widespread anytime soon. And since many websites don't actually care about this level of consistency, it is hard to say whether this will ever become standard.

Kafka

Apache Kafka is a distributed messaging system developed by LinkedIn. It functions similarly to a message queue, but is more specialized for the simultaneous use-cases of real-time and batch log processing. In particular, Kafka emphasizes throughput, scalability, and consumer flexibility over supporting some of the more traditional message queue features, e.g. the Java Message Service (JMS) standard. This paper describes the architecture of Kafka and how it is used at LinkedIn. We'll take a look at a handful of design and implementation decisions in Kafka that enable it to be very efficient and flexible.

At a high level, Kafka follows a messaging model similar to that of other message queues. Producers send messages to "topics," which are essentially namespaces for the messages to live in, and consumers subscribe to the topics that they wish to receive messages from. The first set of implementation details that enable Kafka to be efficient come from within a single server (referred to as a "broker"). Each broker stores some set of partitions for a topic, so that the messages can be distributed among multiple brokers (we'll see how the partitions play into the consumer side later). Each partition can be thought of as a set of messages appended to each other and broken up into files of reasonable size (the paper uses 1GB as the example). Like in many other systems, inserts in Kafka are implemented as appending the message to the end of a file, which is efficient using commodity hardware. The authors, however, choose not to do any in-memory caching of the messages from disk. The reasons are twofold: firstly, the standard OS page cache is quite effective for the access patterns being used (i.e. write-through and read-ahead policies), and secondly, they leverage the Linux sendfile API to send data directly from the file to the socket, avoiding the overhead of having the data live in user space. An interesting effect of this is that the Kafka process itself does not manage a lot of memory, so it is less prone to garbage collection problems and restarts since most of the memory is handled by the OS. These optimizations enable Kafka to have excellent performance on a single broker.

Kafka does interesting things on the consumer side as well. In traditional message queues, the server is typically in charge of keeping track of which messages have been sent to which consumers based on the acknowledgments it has received. Kafka inverts this model and lets consumers request messages from whatever position they wish to. There are a few issues that this can run into. Brokers can no longer clean up messages that they know have been consumed because consumers can always request older ones. This is solved by Kafka having a retention model of a fixed period of time, e.g. 7 days; because of how the partitions are implemented within a single broker, performance does not degrade as partitions grow (unlike with many message queue implementations). This also enables the different patterns of consumption that Kafka was built to handle. Real-time services will consume messages as soon as they appear, while data warehousing and analytics services can, for example, batch consume once per day. The other effect of brokers not controlling message delivery is that consumers who wish to distribute the messages of a topic among themselves (so that each message is consumed once) must coordinate in some way. Kafka achieves this by using the existing highly-available service ZooKeeper. Both consumers and brokers must talk to ZooKeeper in order to coordinate which consumers are responsible for which partitions of a topic and what offset each partition is currently at. So the logic for consumption progress is external to the implementation of a single partition, which allows that to be very simple and efficient as described above.

Messaging systems often form the core of a distributed service, so reliability, scalability, and throughput are of the utmost importance. Having had some less-than-ideal experiences with existing message queues, it is refreshing to see a new approach in Kafka. While it does not satisfy the most stringent durability requirements, Kafka makes up for it with excellent performance and the ability to distribute effectively. It is also an example of how we often do not need to build our own locking/consensus services anymore because ZooKeeper provides a flexible API and strong guarantees on both consistency and availability. I anticipate that future messaging systems will make many of the same decisions that Kafka made in order to satisfy growing data volumes and the fundamentally distributed nature of systems.

Erlang and Riak

Today I learned that Riak, a distributed database based on Amazon's Dynamo, is written in Erlang. This is really cool because I personally have some experience with Apache Cassandra, which is another Dynamo-like distributed database written in Java. Specifically, when I worked with Cassandra, I encountered multiple concurrency bugs that were very difficult to track down and sometimes quite insidious. That's not to fault the writers of Cassandra; it is a complex, multithreaded system that has a lot of moving parts, which is exactly the type of code that led me to write about the difficulty of multithreading. Being written in Erlang, however, Riak should be able to escape some of the reliability issues that arise when dealing with shared, mutable state in Java. I am particularly curious if Riak is more stable than Cassandra, but this information seems hard to come by.

I did stumble across this blog post evaluating the choice of Erlang for Riak five years later. The folks at Basho (company behind Riak) seem to be happy with it, and their rationale for choosing Erlang lines up pretty well with my expectations. Two of the more interesting benefits they describe are that "a 'many-small-heaps' approach to garbage collection would make it easier to build systems not suffering from unpredictable pauses in production" and "the ability to inspect and modify a live system at run-time with almost no planning or cost." The former hits home again with my Cassandra experience, as Java garbage collection on a 16GB heap is not nearly as predictable as one would hope, and the latter is certainly valuable for fast-paced startups and the (in)famous "move fast, break things" philosophy. In the end, I am just happy to see that there are startups out there who have sought alternatives to the concurrent programming hell and built some neat systems as a result of it.

Idempotence

One of the core principles of designing good software systems is composability, the idea being that it should be easy to combine and reuse different components. This has all sorts of benefits, but one of the most important is that it becomes possible for programmers to understand a complex codebase as the composition of simple parts and their interactions. Without some degree of composability and modularity, it would not be possible to build reliable systems over the course of years involving hundreds of engineers because the code simply becomes too complex to understand. Fortunately, people have gotten better at doing this, and a key part of making components easy to compose is the concept of idempotence. It is the simple idea that, when possible, functions should be written in such a way that when you call them multiple times in succession with the same arguments the result does not change. For example, putting a value in a set is idempotent while appending to a list is not since in the latter case the list will keep growing with each append.

Here's an example of an interface which is not guaranteed to be idempotent (BadService) and the resulting attempt to deal with two instances of that interface:


Here, the BadServiceManager is responsible for starting and stopping two BadServices. As you can see, it requires some very ugly state-management that only gets worse as you have more and more components to bring together. The natural way that this should happen with idempotence guarantees is as follows:


In terms of local systems, reduced complexity is the primary reason that you would like idempotence. If that is not compelling enough, as soon as we start working with distributed systems the benefits become very clear cut. Consider a remote procedure call (RPC) made by machine A to machine B. Machine A receives a network timeout while making the call and is now in a predicament; did machine B execute the call or not? This is a very difficult question to answer generically, but idempotence can help. In the case of a network timeout, machine A simply retries the RPC until it receives a successful response from machine B because idempotence guarantees that any repeated calls will not affect the state. This greatly simplifies the job of the RPC client and makes it very natural for distributed components to communicate with each other robustly.

Understanding idempotence is an effective way to improve the composability and reliability of code. Unfortunately, it is not always easy to make functions idempotent and the restriction can make interfaces less clean. The benefits, however, manifest themselves very clearly when you do not need to deal with the unpredictability and complexities of calling non-idempotent interfaces, especially in a distributed setting.

Phi Accrual Failure Detector

Failure detectors are an important piece in designing robust distributed systems. Components must be expected to fail, and the rest of the system should either continue functioning properly (ideal) or at the very least degrade gracefully instead of crashing or becoming corrupted. Because of the unreliable nature of communication over networks, however, detecting that a node has failed is a nontrivial task. The phi accrual failure detector is a popular choice for solving this problem, as it provides a good balance of flexibility and adaptability to different network conditions. It is used successfully in several real-world distributed systems, such as Apache Cassandra (see here) and Akka clusters (see here), and also has a Node.js implementation.

There is a formal theory about the consistency and accuracy guarantees of failure detectors which I will quickly outline before explaining phi accrual (to learn more, please see here). A process is said to "suspect" another process if it believes the other process to have failed. A failure detector is strongly complete if "every faulty process is eventually permanently suspected by every non-faulty process," which is to say that once a process fails, at some point all the processes still running will know that fact. Similarly, a failure detector is strongly accurate if "no non-faulty process is suspected after some time." In summary, non-faulty processes should eventually have the "correct" assessment for all other processes regarding whether they have failed or not, which is a pretty sensible guarantee for a failure detector in a distributed system. Naturally, there are two important metrics for determining how effective a failure detector which satisfies these properties is, namely the time it takes to detect a failure and the rate at which it makes mistakes in suspecting processes. These will guide the discussion of why the phi accrual failure detector makes sense.

Firstly, we should understand the concept of accrual failure detectors. In the formal theory, a process only makes binary decisions about other processes: each of them is either suspected of failure or not. But in practice, we can better capture the uncertainty of these judgments by having a continuous value ($\phi(t)$ in this case, a function of time) and a threshold $\Phi$, where we suspect the process if $\phi(t) \ge \Phi$. Because $\Phi$ is a parameter, accrual failure detectors provide flexibility on top of the basic model. Depending on the requirements of the system, different values can be chosen to optimize for quicker detection or reduced false positives. An example provided in the paper demonstrates how this can be useful: consider a job scheduling system with a master and a set of workers, where the master monitors the status of the workers and assigns jobs to them. Instead of having a binary determination of whether a worker is alive or not, consider having two thresholds $\Phi_1 < \Phi_2$. If $\Phi_1$ exceeded, then stop sending new jobs to the worker, and only when $\Phi_2$ is exceeded assume the worker has failed and reassign any pending jobs. Thus the system can minimize both the chance that it reassigns jobs unnecessarily as well as the chance of assigning jobs to a failed worker.

Phi accrual is one implementation of an accrual failure detector. The way it works is quite simple and relies on the processes sending each other heartbeats at a regular interval, e.g. 100 milliseconds. It keeps track of the intervals between heartbeats in a sliding window of time and measures the mean and variance of these samples, building the corresponding normal distribution with cumulative density function $P(x)$. Then define $\phi(t) = -\log_{10}(1 - P(t - t_{last}))$ where $t_{last}$ is the last time at which a heartbeat was received. The value of $\phi$ will increase the longer it has been since the last heartbeat. Furthermore, the algorithm is adaptive to network conditions because of the measurements in the sliding window. If the network becomes slow or unreliable, the resulting mean and variance will increase; as such, there will need to be a longer period for which no heartbeat is received before the process is suspected. The phi accrual model additionally allows for convenient intuition as to the choice of the threshold $\Phi$. Assuming that the sliding window is representative of the real distribution of heartbeat inter-arrival times, a choice of $\Phi = 1$ means that there is about a 10% chance of a false positive, and in general the probability of a false positive is $0.1^{\Phi}$.

One of the hardest parts about building distributed systems is that there is much less "library code" that can be leveraged. Oftentimes, good software engineering is about not reinventing the wheel, and having well-defined, reusable components is essential to scaling a codebase (or multiple). Phi accrual failure detection seems like it could be an important library that distributed systems can plug in without having to worry about the intricacies of the algorithm or the implementation.

Local vs Remote

While reading through the Akka documentation (the actor and remoting library for Scala), I once again came across a reference to the classic distributed systems paper from Sun discussing the difficulty of unifying local and remote programming models. I briefly touched on the content of the paper in a previous post, but it's worth revisiting in the context of concurrent programming and understanding how the actor model works in a distributed system. With Akka actors, like in Erlang, the programming model revolves entirely around message passing and asynchronous processing even for actors running within the same Java Virtual Machine (JVM). This blurs the line between whether you are sending messages to a local actor and a remote actor, but it is arguably done in the "correct" way. So it is interesting to understand what the difference is between the actor model and other attempts at such unification, e.g. remote procedure calls (RPC), which are generally considered to be less than ideal (although still widely used).

Firstly, we need to recap the four points brought up in the paper that supposedly doom a unified local and remote object-oriented model. They are latency, memory access, partial failure, and concurrency. Latency and memory access are fairly straightforward; because the object on which a method is called lives in a different address space on a remote machine, the method call is going to be much slower and memory accesses have to be managed in some meaningful way across multiple machines. Partial failure refers to the fact that there are multiple components that can fail independently of each other (e.g. the network and the remote entity), which can cause the method call to be in an unknown or perhaps even inconsistent state. Lastly, the issue of concurrency arises because there is no shared synchronization system which can mange the order of method calls on an object, so the programmer must be aware of the fact that an object can receive multiple method calls simultaneously. These points can typically be alleviated by a well-designed RPC library and being diligent in distinguishing local and remote invocations, but that is essentially the argument being made, that in practice they are not truly unified.

So then the question is how the actor model is different and why interacting with local and remote actors can be done in a mostly unified way. The problems of latency and concurrency are non-issues in the actor model because of the asynchronous design and the fact that each actor is processing a single message at a time. And assuming that actors have no shared state and communicate exclusively through the immutable messages passed between them, we can avoid confusing memory access semantics as well. Partial failure is, again, the most complex of the issues, and the actor model itself does not prescribe specific guarantees about message processing, such as idempotence, so it is still left to the programmer to figure out how to keep the system consistent. Regardless, we can see that actors, by design, are better suited for a unified local and remote model since they greatly reduce the space of potential problems to worry about. The corollary, however, is that local operations are more complex due to the way in which communicating between actors and sharing state is restricted, but I am not entirely convinced that this is a bad thing in light of the difficulty of multithreaded programming.

The more I learn about actors, the more optimistic I become about their potential to make concurrent programming less of a disaster. The Akka project is really interesting and gives Scala one more feature that is very much lacking in Java; the mere possibility that synchronized blocks and semaphores can be avoided altogether is enough to motivate me to pursue understanding Akka. Given that the future only holds more concurrent and distributed programming, I feel like programmers should be much better educated about the alternatives to traditional, error-prone multithreading.

Dremel Data Model

It is common knowledge that analyzing large datasets efficiently can benefit greatly from column-oriented storage. That is, instead of storing all the data for a single row together like a traditional database would, separate the columns out and store all of the data for each column together (see the Wikipedia link for an example). The benefits of this are twofold: (1) queries that only access a subset of the columns can reduce the amount of data that has to be retrieved, and (2) compression algorithms often perform better on homogeneous data, e.g. a column with many values of the same type. As such, data stores such as Cassandra and HBase have become widely used when dealing with records that have a large number of columns. Given that both of these projects were based on the original ideas of Google's BigTable, it should be no surprise that Google has continued leveraging the power of column-oriented storage to deal with the scale of data that it processes. Dremel is a project that has been used at Google for a number of years now; it takes the idea of column-oriented storage, generalizes it, and then optimizes it in order to perform queries over tens to hundreds of terabytes of data in seconds.

The key to Dremel's performance is how the data is represented and consequently laid out on disk. First of all, they generalize having a bunch of columns per record to a hierarchical structure of nested columns that supports repeated and optional fields. That may sound familiar because it is exactly the format of protocol buffers, which is the language-independent binary representation of data that Google generally uses, so it is a natural choice for the type of data Dremel should support. Column-oriented storage is simple in the normal case of one (potentially optional) value per column per record since you can easily figure out which value corresponds to which record, but it gets trickier when you allow for the full protocol buffer specification. Consider the following example schema (a simplified version of what is presented in the paper):


This represents a web document, which has an ID and a set of names associated with it; each name is a URL that points to that document and the languages associated with the URL. The generalization of column-oriented storage to handle nested columns is that each leaf field in the hierarchy is stored separately. For example, all values of "docId" are stored together, and all values of "name.language.country" are stored together since both are leaf fields in the schema for a document. Let's look at two example records which we will assume are stored in the order presented:


The "docId" field is simple and only needs to store the values "10" and "20" since it is a top-level, required field. The "name.language.country" field, on the other hand, will need to store the values "us", NULL, NULL, "gb", NULL. We will see why these NULL values are necessarily shortly, but they are basically placeholders for potential values of "name.language.country."

It should be clear that because of the repeated and optional values we cannot only store the values above without additional metadata, since doing so would lead to the loss of record boundaries. To solve this, Dremel introduces two metadata fields known as the "repetition level" and the "definition level." These are logically attached to every column value in order to preserve record boundary information. The repetition level handles repeated fields and indicates the depth in the hierarchy that was repeated between the current column value and the previous one (0 is the root, meaning a new document). The definition level handles optional fields and indicates the depth in the hierarchy that actually exists for the current value; this is only relevant for NULL values. For the "name.language.country" field, Dremel would store the following:

#	value	rep	def
1	"us"	0	3
2	NULL	2	2
3	NULL	1	1
4	"gb"	1	3
5	NULL	0	1

In row 1 of the table, rep = 0 because it is a new record, and def = 3 because all 3 levels of the hierarchy ("name", "language", and "country") are defined, which is why we have a non-NULL value. In row 2, rep = 2 because we are repeating "name.language" when going from the previous row to this one, and def = 2 because only "name" and "language" are defined so the value is NULL. In row 3, rep = 1 because we repeated at the "name" level, and def = 1 because only "name" is defined. In row 4, rep = 1 because we again repeated at the "name" level, and def = 3 because we have a real value. Lastly, row 5 has rep = 0 because it is a new record, and def = 1 because only "name" is defined. The meanings of the repetition and definition levels can be confusing at first, but if you understand how they are derived in the above table, you should be able to convince yourself that it is a lossless representation. Dremel then optimizes to the bit-level how much data they need to actually store, such as omitting representation and definition levels whenever possible and using as few bits as necessary to store the values.

It should be no surprise that Google's solution for interactive queries on huge datasets involves minimizing the amount of data that needs to be retrieved from disk, a concept that database engineers have employed for decades. Since the volume of data being collected today far outpaces amount that can be read off of disk in a few seconds, there seem to be few options in the analytics space for efficient queries other than designing to reduce what needs to be read. But one can certainly imagine more complex queries that involve a large number of fields or even joining two fields that would be extremely valuable yet essentially impossible given the limitations of today's technology. This area seems ripe for additional work in pushing the boundaries of what types of queries are possible on the huge datasets that companies are collecting, but Dremel is definitely a solid approach that Google has derived a lot of value from.

Spanner TrueTime

Spanner is one of Google's recent developments in the area of large-scale distributed systems, and its design from the very beginning was intended for deployment at the global scale. Not only that, but because many consumers of systems similar to Spanner (e.g. BigTable and Megastore) wanted better consistency guarantees, the authors also built externally-consistent distributed transactions. This level of consistency is quite rare among the popular NoSQL databases out there, with many of them opting for eventual consistency, so it is quite refreshing. Moreover, Spanner uses schemas which consist of "semi-relational tables" (they require primary keys) and provides a "SQL-based query language," which is interesting given the direction that other distributed databases are generally heading. The technical paper for Spanner can be found here.

What appears to be the most revolutionary concept behind Spanner is how it deals with time. Managing time in distributed systems is a well-known difficulty and has been for many decades (see Leslie Lamport's paper), so it might not be surprising that the combination of strong consistency and global scale ends up focusing a great deal on time. Google's approach to solving this problem has two parts; the first is formalizing time uncertainty and working with it, rather than around it. They introduce TrueTime, which is an API that the machines in their system use to quantify their own uncertainty regarding the time. It has just three methods, which are TT.now(), TT.after(), and TT.before(). The first returns an interval that specifies the earliest and latest times that it could be right now, while the latter two allow you to query whether it is definitely after or before a point in time.

It turns out that by using TrueTime and a lot of Paxos, Spanner is able to guarantee a level of consistency that is stronger than that of many other distributed systems while still being very efficient. For example, you can implement snapshot isolation for read-only transactions and point-in-time reads by ensuring that all replicas can "safely" read at a certain timestamp (i.e. a later timestamp has been committed). And during transactional writes, the commit is delayed so that it occurs at a time known to be after the recorded timestamp, as specified by TrueTime, which prevents out-of-order writes. The details of how the authors manage all of this can be the found in the paper, but suffice it to say that TrueTime on top of Paxos is what allows all of these consistency guarantees to be made.

Lastly, it should be pointed out that with the Spanner design, if the clock skew of the individual machines is high, performance will suffer greatly. Commits on read-write transactions will be forced to block for long periods and thus limit latency and throughput. And so the second part of Spanner time management is in the hardware: GPS and atomic clocks. By using clocks that give you much better performance in terms of skew along with a custom clock synchronization protocol (instead of NTP), Spanner is able to maintain very low levels of uncertainty about the time. Their novel approach to time accounts for uncertainty so that they can reason formally about consistency but heavily leverages clock accuracy for performance. The fact that Google is able to deploy this type of system globally (it is powering their new advertising backend) speaks a lot to the affordability of accurate clocks today.

All in all, Spanner is a unique distributed system that brings together a set of ideas and even some hardware to solve a very hard problem. It will be interesting to see if the allure of a high-performance system that provides strong consistency starts drawing the attention of the NoSQL community and whether the TrueTime API becomes more common as a way to manage time uncertainty.

Distributed Matrix Computation

Anyone who has worked with MapReduce knows that, while it suits many distributed computations well, it is very difficult to implement matrix computations using the framework. It is possible to represent matrix-based algorithms such as PageRank using MapReduce (e.g. here) but it definitely isn't the most natural or efficient way. Moreover, PageRank is pretty simple as far as matrix computations go, and once you want to do things like Cholesky decompositions for solving systems of linear equations or singular value decompositions it gets very tricky. MapReduce was simply not built for these use cases, so there is nothing to fault it for, but since it is the standard go-to distributed computing framework people have tried to do just about everything with it. Different problems require different solutions, though, and the creators of MadLINQ, a system coming out of Microsoft Research, are trying to overcome the difficulty of implementing complex, distributed matrix algorithms with their new system.

The MadLINQ system shares a lot in common with its cousin DryadLINQ that it integrates with. They both have use the directed acyclic graph (DAG) model of computation where each node represents some unit of computation that depends on the outputs of some other nodes. For example, MapReduce itself can be represented in this model by a set of "map" nodes and a set of "reduce" nodes where an edge exists from each of the map nodes to each of the reduce nodes. The key then lies in how MadLINQ maps a matrix computation into a DAG that can be efficiently executed in parallel. To do this, the authors use the familiar tiling abstraction applied to many parallel matrix operations (e.g. tiled matrix multiplication) where the matrix is broken into disjoint, square sub-matrices to operate on. Each node in the DAG is then an optimized, parallel matrix operation on a set of input tiles from the original matrix or intermediate results. One nice thing is that they abstract away the underlying matrix library for the individual node computations so that improvements in that space (which many people work on) will result in improvements in MadLINQ.

One interesting optimization implemented in MadLINQ is what the authors call "fine-grained pipelining." In both MapReduce and DryadLINQ, a node in the DAG must wait for its predecessors to completely finish processing before reading all of their output and executing. This has two negative consequences: first is the fact that during periods where there is low parallelism in the DAG (you can only break down the units of execution so far before the overhead exceeds the benefits) the cluster will be idle, and second is the fact that network utilization will be more bursty because a node decides to read the entirety of its inputs at once. MadLINQ allows individual nodes to produce partial output and consume partial input, enabling pipelining at the sub-node level (hence the term "fine-grained"). This is particularly useful in the context of matrix computations because many of the individual nodes also execute tile-based algorithms which lend themselves to this granular pipelining. The authors went through quite a bit of trouble to optimize the system as much as possible to really "solve" the use case of distributed matrix computations.

MadLINQ really speaks to how hard problems in distributed systems are. MapReduce has been a huge step forward in the last 10 years, but it has its limitations, and as more and more people and companies are turning to large-scale computations and processing, they will need to invent new solutions to solve the wide array of problems out there. The fundamental design decisions that must be made when building a new distributed system inherently limit its applications in a way that, for example, modern processors are not limited. People (generally) do not design specific processors for their applications anymore, and one would expect distributed systems to eventually reach that point as well, but there is certainly a long road ahead.

Scalding

A couple of days ago, I went to a talk hosted by the Bay Area Scala Enthusiasts on the topic of "Scalable and Flexible Machine Learning with Scala." The talk primarily focused on using Scala with Hadoop to very easily run machine learning algorithms like PageRank, k-means clustering, and decision trees. It was interesting to hear about how the speakers (from LinkedIn and eBay) used machine learning in practice and why they feel Scala is particularly suitable for the task in the context of today's libraries and frameworks. In particular, they showed an impressive library called Scalding that came out of Twitter which is designed to make writing Hadoop workflows in Scala very natural.

Let's start with a quick code snippet taken from the Scalding website.


This piece of code is the implementation of word count, the prototypical MapReduce example, on Hadoop. Besides being impressively short, it is very close to Scala Collections syntax which means it comes naturally to anyone who has programmed in Scala extensively. The idea is to be able to take code the operates on collections and easily translate it into a MapReduce job that can run on Hadoop, which makes a lot of sense since MapReduce is essentially based on the functional programming paradigm.

Of course, it's important that the resulting MapReduce job be efficient. The first two method calls, flatMap and groupBy, don't actually do any work themselves; instead, they simply create flows, an abstraction from the Cascading library which Scalding is built on top of. Much like other languages which attempt to hide the details of MapReduce, Cascading and Scalding construct job/execution flows which are only optimized and executed when you decide to write some output. This is essential because any sort of on-demand execution of map and reduce steps would necessarily lead to inefficiencies in the overall job (e.g. multiple map steps in a row would not be combined and executed on the input all at once).

So why Scalding over something like Pig, which is a dedicated language for Hadoop? One major reason is that it is a huge pain to use custom methods in Pig because you have to write them outside of the language as it is not expressive enough. If you integrate Hadoop into Scala, then you get all of Scala and whatever dependencies you want for free. Furthermore, Pig does not have a compilation step, so it is very easy to get type errors somewhere in the middle of your MapReduce job, whereas with Scalding you would catch them before running the job. In practice, this is a big deal because MapReduce jobs take nontrivial amounts of time and compute resources. Pig has its uses, but Scalding brings a lot to the table in the areas where Pig is lacking.

Running Hadoop jobs is becoming very easy, and Scalding is a great example of how you can implement moderately complex machine learning algorithms that run on clusters with just a handful of lines of code. If there exists a Java library that does most of the hard work, even better; you can integrate it right into your Scalding code. While you probably still have to understand Hadoop pretty well to get the optimal performance out of your cluster, people who are less concerned about resource utilization or who don't have absurdly large datasets benefit greatly from abstractions on top of Hadoop. And everyone wins when it is easy to write testable, robust code even when you are not an expert. This is certainly the direction for many distributed computing and machine learning use cases going forward, and it will be really cool to see what new technologies are developed to meet the demand.

The slides for the talk can be found here.

Erasure Coding

Durability is a critical feature of distributed storage systems. Because these storage systems are typically built on a large number of machines using commodity hardware, individual component failures are expected and common; thus durability comes from how fault tolerance is designed into the system. The original Google File System (GFS) was among the earliest systems built at a scale where these problems started to manifest, and one of the key features was that all data in GFS was redundantly stored in three copies. Combined with the proper distribution of these three copies so they don't fail simultaneously, this design allowed Google to scale out their storage system without losing data. While GFS was quite a landmark piece of technology, we've come a long way since then, and people are constantly devising new ways to achieve durability while optimizing other important features of distributed storage systems like latency, storage overhead, and reconstruction cost.

Storing identical copies of data is a simple strategy for redundancy, but more involved techniques such as erasure coding can greatly improve the storage overhead while providing the same durability guarantees. The idea behind erasure coding is to encode a message of $K$ symbols using $N > K$ symbols such that you can reconstruct the original message using only a subset of the $N$ symbols. For example, you can think of storing three copies of data as erasure coding with $K = 1$ and $N = 3$, where as long as you have one of the three copies you can reconstruct the original data. Another common example is encoding $K$ bits using $N = K+1$ bits where the extra bit stores the parity/XOR of the first $K$ bits; losing any single bit will still allow you to recover the original message. These examples demonstrate how erasure codes are redundant and prevent losing the message even when individual symbols are lost (e.g. a machine storing data dies).

The most common erasure codes are Reed-Solomon codes, which are present in CDs, DVDs, and barcodes to ensure that these media are resistant to partial damage. Reed-Solomon codes extend the idea of the parity bit by leveraging polynomial interpolation. Given the message of $K$ symbols, consider the polynomial of degree (at most) $K-1$ with the symbols as the coefficients. Evaluate the polynomial at $N > K$ points known to both the sender and the receiver, and send the results of those evaluations. As long as $K$ of the resulting symbols are received, the polynomial can be interpolated and thus the original message reconstructed. The details of the algebra on polynomial rings and an efficient decoding algorithm are beyond the scope of this post, but suffice it to say that people have figured out how to use Reed-Solomon codes very effectively. It shouldn't come as a surprise then that the new GFS (discussed here) uses them and the Hadoop File System (HDFS) has a module which adds the functionality.

The story doesn't end there, though. Last summer, a paper came out of Microsoft Research and the Windows Azure Storage (WAS) team that describes a new technique called local reconstruction codes (LRC) used in WAS, Microsoft's equivalent of GFS and HDFS. One of the downsides of Reed-Solomon codes in storage systems is that, if one of your original $K$ symbols is lost, you must read a full $K$ other symbols in order to recover that one symbol. When these symbols are chunks distributed across machines, it can be costly to do that many accesses. LRC can be thought of as an extension of Reed-Solomon codes which adds an additional parameter that determines how many local groups there are within a message. It facilitates the reconstruction of lost symbols by leveraging additional redundancy introduced at the local group level. The work demonstrates that the technique is more efficient both in terms of storage overhead and reconstruction cost than Reed-Solomon codes when both methods are required to provide the same fault tolerance as storing three copies of the data (which is still the standard for systems today).

The theory behind LRC is really cool (and my two-sentence summary doesn't do it justice, so read the paper if you're at all interested), but the additional practical discussions in the paper are equally interesting. Some of the key points that I found particularly enlightening were the following:

• When retrieving a fragment (unit of a file which is erasure coded) from WAS, a request to reconstruct the fragment is issued simultaneously, so that if the fragment lives on a machine that is under heavy load it's possible that the reconstruction is more efficient and returns first. I imagine that this has a nice effect on the long tail of retrieval times.
• As files are being written, they are initially stored as three redundant copies, but once a file is "sealed" and can no longer be written to, the system will asynchronously erasure code it and remove the redundant copies. This is so the erasure coding doesn't have any performance impact on the "critical path of client writes."
• Even the process of erasure coding must be fault tolerant. The WAS system stores progress information for files that are being erasure coded so that fragments which are complete don't need to be encoded again after a failure. Furthermore, they employ many cyclic redundancy checks during the process and perform extensive randomized reconstruction tests prior to committing to the erasure-coded version and deleting the original copies.

Erasure coding is a powerful technique for providing durability in distributed storage systems and seems to be becoming the standard. LRC is a neat innovation in this space, and it will be exciting to see how these technologies evolve as these types of systems become more and more prevalent.

Concurrency with REST

Representational State Transfer (REST) is one of the main architectural models that web services use to communicate with each other. While it is not tied to HTTP, it is most often used on top of HTTP for various web APIs (e.g. Amazon S3, Stripe, Netflix, Zendesk, Greendizer). A very common pattern when writing REST APIs is the management of entities through the four operations of create, read, update, and delete (CRUD). For example, these entities could be buckets in Amazon S3, customers in Stripe, or tickets in Zendesk. The CRUD operations map very naturally to HTTP verbs (POST, GET, PUT, and DELETE, respectively), so you end up with an interface that is clean and easy to understand for developers integrating with your service.

One problem that inevitably arises because we are working in a distributed setting is how to deal with concurrency. What kind of semantics should be guaranteed for clients of the API who are concurrently operating on the same entities? Reads are innocuous here because they don't modify state; creates and deletes are not particularly worrisome either as they can only happen once each. This leaves us with just the update (i.e. conventional write) operation to think about. It's easy to see how two unsynchronized updates can clobber each other in ways that are undesirable, leading to the lost update problem. But remember that HTTP is stateless, so locking a resource is quite dangerous when the client who obtains the lock does not successfully issue another request to release it. If we can't lock, then how do we solve this problem? Enter optimistic concurrency control.

Optimistic concurrency control is based on the assumption that in most cases there is not contention for a resource (hence the "optimistic"). It relies on versioning the entities in the system such that every time an update is made to an entity its version changes, e.g. a counter that increments on each update. When a client gets an entity, he is also given the current version information for that entity; if he then wishes to update the entity, he must provide the same version information that was received. Before performing the update, the server checks to make sure that the version provided by the client matches the version on the server (i.e. nobody else has performed an update) and only proceeds with the update if that is the case. These simple steps ensure that no updates are lost and that the system is robust to client failure; the performance is also good provided the resource is not highly contended.

HTTP has a built-in feature for this version information called the ETag header which is used for caching and conditional requests. When a client gets an entity from the server, the ETag header in the response will be populated with the version information. In practice, the version information is typically the result of applying a cryptographic hash function to either the data itself or some unique version identifier. If the client subsequently decides to update that entity, he should provide the version information in an If-Match header in the update request (a conditional PUT). This allows the server to do proper validation of the client's update as per the optimistic locking protocol and return the appropriate status code if the version no longer matches. If the client gets a response that indicates a version mismatch, he can simply get the entity again and perform the update as desired, the important point being that the other update is acknowledged.

The use of the ETag header for concurrency control is not necessary in all REST APIs, but it can be an easy way to relieve clients of worrying about complex synchronization themselves.