Pacemaking

The problem

Imagine there's some number of servers $N$ in a fully-connected P2P network. These servers can copy files from each other. In this imaginary system, we have a single machine that every $δ$ minutes produces a file that all other machines need to get a hold of (say, for example, a database checkpoint), and this file happens to be relatively large, let's say > 200GB for the sake of argument.

Consumers of the file would need to learn about the file's existence, and then download it. We could imagine a push notification system, or some sort of polling system, but for the sake of simplicity let's assume the producer stores a record of the existence of this file in a shared location. Therefore, let's say every $1 m$ , each server will check this shared location to see if there's a new file to download. If there is, it will download it from any of the machines that already have it. If there isn't, it will do nothing.

The challenge

This kind of consumption pattern is prone to denial of service attacks on the producer of the file: if the producer is the only machine that has the file, then many machines are likely to try to download the file from the producer at the same time, which could easily overwhelm the producer. This is a classic example of a "hot spot" in a distributed system.

One thing to notice about this system is that it will almost always wind up stressing the producer:

Machines are typically (due to deployment cycles) started up roughly at the same time, which means they will tend to come against the producer in groups.
A subset of machines may be enough to overwhelm the producer (and in practice it often is). Once the producer is overwhelmed and slow, more machines will pile on, making the problem worse. This is a positive feedback loop that can easily lead to a denial of service attack.
An alternative here is something in the form of a "retry storm": the producer is slow, triggers timeouts (or straight up failures), and the machines retry, making the problem worse.

This is not a happy situation to have if you want to sleep soundly at night (ask me how I know!). So, what can we do about it?

What we're really dealing with is a thundering herd: a single producer about to get stampeded by every consumer at once. Classic load balancing doesn't apply since there's only one server with the file at the start. Reactive defences like rate limits, queues, or backoff only kick in once the stampede has started, by which point the producer is already overwhelmed.

The thing to notice is that consumers double as future sources: once a machine has the file, it can serve it to its peers over the same P2P fabric. So if we can stagger the herd into waves, the replica set grows as each wave completes, and later waves can pull from peers instead of from the producer.

There's plenty of ways to do this, for example, the producer could proactively send the file to future consumers before advertising it (guaranteeing there's a pool of machines in existence by the time it's advertised), or it could advertise the file only to a subset of machines, and then have those machines advertise the file to other machines, etc.

The issue with most of these ideas is that they are relatively costly in terms of implementation (for example, we now need to implement this thing which allows us to proactively send files!), and we all know no one likes large PRs with high maintenance cost. We already have a content distribution network in place, and a place to advertise the file's existence, so can we make do?

The key idea behind pacemaking is this: we can manipulate the chance of any given machine downloading a file in any given time period in order to ensure the load on the producer dissipates over time by adding more replicas in a controlled fashion. Let's dig in.

Distributing downloads over time

We have $N$ servers that we want to distribute the file to over time. One way to distribute the load over time is to arrange downloads in a way that machines downloading the file at any given time are roughly the same numbers as the machines that already have the file. To do so, imagine we split the time into buckets:

At $T = 0$ , we have exactly 1 machine (the producer) who has the file, so we want at most 1 machine to download the file.
At $T = 1$ (assuming everything went well) we have 2 machines who have the file, so we want at most 2 machines to download the file.
At $T = 2$ (assuming everything went well), we have 4 machines who have the file, so we want at most 4 machines to download the file.

So essentially, at the beginning of the $T$ -th bucket of time, we want at most $2^{T}$ machines to download the file, as $2^{max (T - 1, 1)}$ machines already have downloaded the file by that point in time. At the end of the $T$ -th bucket of time, $2^{T + 1}$ machines will have downloaded the file. Since we have $N$ machines that need to download the file, we need $⌈ lo g_{2} (N)⌉$ buckets over time to ensure all machines have downloaded the file; and the last bucket will have $N - 2^{⌊ l o g_{2} (N)⌋}$ machines downloading the file.

A plot of that looks something like this:

A few things are worthy of notice here:

We do not actually have a way to distribute downloads over time such that they fit this distribution.
We are assuming all file downloads are successful.
We have conveniently assumed that the duration of the file download fits within the bucket, so that by the end of the bucket the machines will have downloaded the file.
The last bucket is bound to have the highest number of concurrent downloads, so most machines will have to wait essentially $(B - 1) S$ until they can download the file, where $B$ is the number of time buckets and $S$ is their size.

We'll work through these details in the next sections.

Allocating machines to time buckets

In order to achieve this distribution of downloads, we would like to allocate machines to buckets of time such that the number of machines allocated to each bucket is the same as the number of machines that have downloaded the file by that point in time. One way to do this is to sample from a distribution that has this property.

So, imagine $X$ is a discrete random variable that represents a bucket number, with support over ${0, 1, \dots, B - 1}$ , where $B$ is the bucket number. We would really like to ensure something similar to $P (X = i + 1) = 2 P (X = i)$ .

One way to do so is to ask $P (X = i) = 2^{i} q$ (which would trivially satisfy the property we'd really like to have), then we can solve for $q$ , since it must be a probability distribution:

i = 0 \sum B - 1 P (X = i) i = 0 \sum B - 1 2^{i} q q i = 0 \sum B - 1 2^{i} q (2^{B} - 1) q = 1 = 1 = 1 = 1 = \frac{1}{2 ^{B} - 1}

Which leaves us with $P (X = i) = \frac{2 ^{i}}{2 ^{B} - 1}$ . Now, we have some fancy math which tells us how to distribute machines over buckets, only now we need to sample from this distribution somehow.

Inverse transform sampling is a method to sample from a distribution given its cumulative distribution function (CDF). The CDF of this distribution is given by:

F (x) = P (X \leq x) = i = 0 \sum x P (X = i) = i = 0 \sum x \frac{2 ^{i}}{2 ^{B} - 1} = \frac{2 ^{x + 1} - 1}{2 ^{B} - 1}

and the inverse of that function is:

F^{- 1} (y) = argmin_{x} (\frac{2 ^{x + 1} - 1}{2 ^{B} - 1} \geq y) = argmin_{x} (2^{x + 1} - 1 \geq y (2^{B} - 1)) = argmin_{x} (2^{x + 1} \geq y (2^{B} - 1) + 1) = argmin_{x} (x + 1 \geq lo g_{2} (y (2^{B} - 1) + 1)) = ⌊ lo g_{2} (y (2^{B} - 1) + 1)⌋

Then, inverse transform sampling tells us to:

Sample $y$ from a uniform distribution between 0 and 1.
Compute $x = ⌊ lo g_{2} (y (2^{B} - 1) + 1)⌋$ .

This will give us a sample bucket $x$ from the distribution we want, which looks like this:

At last, the algorithm

This algorithm has two sides to it: the producer of the file, and the consumer of the file. On the producer's end, we will keep everything as-is, but we'll add some metadata to the file:

The file creation time $T_{current}$ .
The expected time of creation for the next file $T_{next}$ , if it's known.

On the consumer's end, the key thing is this: we might poll for the file multiple times before a new file is created, so we need to ensure that whatever bucket we're in, it's consistent over time.

So, we know that:

We have $N$ machines that need to download the data.
They need to be distributed over $B = ⌈ lo g_{2} (N)⌉$ buckets.
We need to sample from the distribution we derived above to determine which bucket we're in.
We need to ensure that the bucket we're in is consistent over time every time we poll until the next file is created.
We have $δ = T_{next} - T_{current}$ until the next file is created.

The missing piece here is determining the size $S$ of the buckets. This can be done in any number of ways, such as $S = \frac{\frac{1}{2} δ}{B}$ , or via some sort of estimate of the download time. The important part is that $T_{current} + S B < T_{next}$ , so all machines are done downloading before the next file is available.

Now that we have all this, we need to determine which bucket we're in. The problem is that if we follow the inverse transform sampling method, we will get a different bucket every time we poll for a new file. This is not what we want, as we want the machine to be assigned the same bucket every single time it checks, at least until a new file is available to download.

One easy way to do this is to use hashing. We can hash the file creation time and the machine name to get a number that's roughly uniformly distributed (as in $r = hash (fileCreationTime, machineName)$ ), and then use that number to determine the bucket by setting $b = F^{- 1} (r)$ . This way, the machine will always be assigned the same bucket until a new file is created.

The rest of the algorithm is straightforward. Every time we check for a new file, we do the following:

If there is no new file, we do nothing.
Otherwise, generate $r$ , and sample the current bucket $b = F^{- 1} (r)$ . We proceed to download the file only if $T_{next} \geq now \geq T_{current} + S b$

What was thrown under the rug

As per above, we assumed that all file downloads are successful, and that the download time fits within the bucket. In my particular production scenario, I had $1 m$ buckets with $δ = 15 m$ , and downloads would certainly fit within that bucket, so I didn't have to worry about it. If you have a different scenario, you might need to adjust the bucket size accordingly.

The other issue is that we assumed all file downloads are successful. This is a bit of a problem, as we don't have a way to ensure that all file downloads are successful. Due to particular details of the production system I was working with, this didn't actually matter because the probability of a file download failing was in practice very low. This can, however, be somewhat accounted for by modifying the distribution to account for the probability of a file download failing, and then sampling from that renewed distribution.

Conclusion

This algorithm is a simple way to ensure that the load on a producer of a file is distributed over time. It's obviously not perfect and has a number of underlying assumptions, but it does have the benefit that it can be implemented in like 10 lines of code and doesn't require much infrastructure. It's a neat trick to have in your back pocket if you ever find yourself in a situation where you need to distribute load over time, or more likely as inspiration if you ever find yourself in a similar situation.

I have run this algorithm in production and it has worked without any issues over the course of what is now several years (and in fact, the smoothing it introduced eliminated several ways in which the system used to fail).