9. Counting across replicas

A GPU platform counts all day. It counts the GPU time each team’s jobs have used, so that the time can be billed or shared out fairly, and it counts the GPUs each team has in use right now, so that a quota can be enforced. Inside one cluster, counting is not hard: the API server is the single place every write passes through. A platform that spans several sites has several places where things happen, and they cannot always reach each other.

In this chapter, we will count across three replicas of a usage service, one beside each group of machines in our fleet. We will first see why the obvious ways of keeping the replicas in step lose counts or count twice. Then we will write counter, in Go and in Rust, which builds a count that survives a network partition, extends it to a count that goes down as well as up, and finally shows the one thing such a count cannot do, which is to enforce a limit, and what does that job in its place.

One authority, or several

Suppose each group of machines in our fleet has a small service beside it that records usage: one on the shelf with the Thors, one on the desk with the Sparks, one on the bench with the Orin NX. Jobs report to the nearest one. Each of these is a replica of the usage service, and all three are meant to hold the same totals.

The first design gives the count a single home. One replica, or a database that all three write to, holds the number, and every report becomes a write to it. etcd is a database of this kind: it stores each write on a majority of its members before it acknowledges the write, so that all members agree on the order of writes. The count is always exact.

The cost appears when the network between the sites fails. Our three groups are joined only by the house network. If it goes down, the bench is alone. A replica that cannot reach a majority has two choices, and both are bad for a usage counter:

• It can refuse reports until the network returns. The jobs near it keep running, and their usage for that period is not recorded.
• It can accept reports into a count of its own. Now there are two counts, and the design gave us no rule for combining them afterwards.

A split like this is called a network partition, and the choice is a general one. During a partition, a replica that is cut off can stay consistent, by refusing work it cannot coordinate, or stay available, by doing the work alone. It cannot do both. For a count of usage we want availability: a report should be accepted wherever it arrives. So we need the thing the second choice lacked, a rule for combining counts that were kept apart.

Two rules that do not work

Let each replica keep one number, its total. After some jobs have reported, the shelf holds 300 GPU-seconds, the desk 120, and the bench 40. The true total is 460. Now the replicas exchange numbers and each has to decide what its total should be.

Keep the larger. Each replica replaces its number with the larger of its own and the one it received. All three end at 300. The 160 GPU-seconds recorded at the desk and the bench are gone, because the rule treats two different counts as two versions of the same count.

Add the other’s number. Each replica adds what it receives to its own. After the shelf and the desk exchange once, both say 420, which is right for those two. But exchanges repeat, and messages are sometimes delivered twice. When the desk’s 120 arrives at the shelf a second time, the shelf says 540. A rule based on adding counts every report again each time the replicas talk.

Both rules fail for the same reason. A single number does not say whose reports it contains, so a replica that receives one cannot tell which part of it is news.

One entry per replica

The fix is to keep that information. Each replica holds a row with one entry for every replica, itself included. Two rules govern the row:

• A replica records a report by adding to its own entry. It leaves every other entry as it received it.
• When a replica receives another’s row, it goes through the entries one by one and keeps the larger of its own value and the received one.

The count is the sum of the row.

“Keep the larger” failed for totals and works here, and the difference is what the two numbers being compared mean. An entry is written by exactly one replica and only ever increases. So when two values of the same entry differ, they really are an older and a newer version of the same thing, and the larger is the newer.

Let us run the earlier example. After the first reports, written as rows of (shelf, desk, bench):

• the shelf holds (300, 0, 0),
• the desk holds (0, 120, 0),
• the bench holds (0, 0, 40).

The shelf receives the desk’s row and takes the larger in each position: (300, 120, 0), which sums to 420. If the same row arrives again, nothing changes, because 120 is not larger than 120. When the bench’s row arrives, the shelf holds (300, 120, 40), and the sum is 460.

This structure is called a grow-only counter, or G-counter. Figure 9.1 follows three of them through a partition.

What the merge guarantees

The merge has three properties, and each one removes a requirement from the network between the replicas.

• Merging the same row twice gives the same result as merging it once. So a message may be delivered twice, and replicas may exchange rows as often as they like.
• Merging row A and then row B gives the same result as B and then A. So messages may arrive in any order.
• Merging A with the result of merging B and C gives the same result as merging A and B first and then C. So a row may travel through other replicas on its way, picking up their entries, and nothing is lost or doubled.

Together they mean that any two replicas that have received the same reports, by any route, hold the same row. Nothing has to be coordinated, and a replica that was cut off for an hour catches up with one exchange. Data structures with a merge of this kind are called conflict-free replicated data types (CRDTs).

Counting down

A count of GPUs in use has to go down when a job ends, and “keep the larger” cannot represent a decrease: a replica that lowered its entry would have the old, larger value merged straight back in.

The way round this is to use two grow-only counters. One counts every increase and the other counts every decrease, and the value is the first minus the second. Both halves only grow, so both merge safely. This is a PN-counter, for positive and negative.

What a merged count cannot do

A quota is a limit: this team may use at most six GPUs at once. Enforcing it means refusing a job when the count is at the limit, and here the merged count lets us down.

Suppose the shelf and the desk both know that four of the six GPUs are in use, and then lose contact. A two-GPU job arrives at each. Each replica looks at its own count, sees four in use, and admits the job, correctly by everything it knows. When contact returns and the counts merge, eight GPUs are in use against a quota of six.

No cleverer merge fixes this. A limit is a statement about the sum of what all replicas have done, and a replica that is cut off cannot know what the others are doing at this moment. What we can do is divide the limit beforehand, while the replicas can still talk. Each replica is given a share of the quota and admits jobs only out of its own share. The shares add up to the quota, so whatever the replicas do while apart, the total stays within it. When they are in contact, a replica with unused share can hand some to one that has run out.

The price is that a job can be refused while capacity exists. If the shelf has one GPU of its share left and the desk has one, a two-GPU job is refused at both, even though two GPUs are free in total, until the replicas can talk and move a share across. The partition forces a choice here too, and for a limit we chose to refuse.

Building counter

The grow-only counter

A GCounter is an array with one entry per replica. Add touches one entry, Merge takes the larger value of each, and Value sums them.

Listing 9.1 The grow-only counter, in Go. code/go/counter/main.go

// GCounter is a count that only grows. It keeps one entry per replica, and
// each replica adds only to its own entry.
type GCounter [replicas]int

// Add records n more at the given replica.
func (c *GCounter) Add(replica, n int) { c[replica] += n }

// Merge takes, entry by entry, the larger of this counter's value and the
// other's. Merging is safe to repeat and to do in any order.
func (c *GCounter) Merge(other GCounter) {
	for i := range c {
		c[i] = max(c[i], other[i])
	}
}

// Value is the count: the sum of every replica's entry.
func (c GCounter) Value() int {
	total := 0

	for _, n := range c {
		total += n
	}

	return total
}

Listing 9.2 The grow-only counter, in Rust. code/rust/counter/src/main.rs

/// A count that only grows. It keeps one entry per replica, and each replica
/// adds only to its own entry.
#[derive(Clone, Copy, Default)]
struct GCounter([i64; REPLICAS]);

impl GCounter {
    /// Records n more at the given replica.
    fn add(&mut self, replica: usize, n: i64) {
        self.0[replica] += n;
    }

    /// Takes, entry by entry, the larger of this counter's value and the
    /// other's. Merging is safe to repeat and to do in any order.
    fn merge(&mut self, other: GCounter) {
        for (mine, theirs) in self.0.iter_mut().zip(other.0) {
            *mine = (*mine).max(theirs);
        }
    }

    /// The count: the sum of every replica's entry.
    fn value(self) -> i64 {
        self.0.iter().sum()
    }
}

The counter is a fixed-size array in both languages, which makes it a plain value: assigning it copies it, and “send my row to another replica” in the program is passing the array to Merge.

The counter that goes down

PNCounter holds two grow-only counters. A positive change goes into Up, a negative one into Down with its sign removed.

Listing 9.3 The up-and-down counter, in Go. code/go/counter/main.go

// PNCounter is a count that goes up and down, built from two counters that
// only grow: one for the increments and one for the decrements.
type PNCounter struct {
	Up, Down GCounter
}

// Add records a change of n, which may be negative, at the given replica.
func (c *PNCounter) Add(replica, n int) {
	if n >= 0 {
		c.Up.Add(replica, n)
	} else {
		c.Down.Add(replica, -n)
	}
}

// Merge merges both halves.
func (c *PNCounter) Merge(other PNCounter) {
	c.Up.Merge(other.Up)
	c.Down.Merge(other.Down)
}

// Value is the increments minus the decrements.
func (c PNCounter) Value() int { return c.Up.Value() - c.Down.Value() }

Listing 9.4 The up-and-down counter, in Rust. code/rust/counter/src/main.rs

/// A count that goes up and down, built from two counters that only grow: one
/// for the increments and one for the decrements.
#[derive(Clone, Copy, Default)]
struct PnCounter {
    up: GCounter,
    down: GCounter,
}

impl PnCounter {
    /// Records a change of n, which may be negative, at the given replica.
    fn add(&mut self, replica: usize, n: i64) {
        if n >= 0 {
            self.up.add(replica, n);
        } else {
            self.down.add(replica, -n);
        }
    }

    /// Merges both halves.
    fn merge(&mut self, other: PnCounter) {
        self.up.merge(other.up);
        self.down.merge(other.down);
    }

    /// The increments minus the decrements.
    fn value(self) -> i64 {
        self.up.value() - self.down.value()
    }
}

One number per replica

The first experiment is the rule that fails: one total per replica, synchronised by keeping the larger.

Listing 9.5 Keeping the larger of two totals, in Go. code/go/counter/main.go

// oneNumber counts with a single number per replica and synchronises by
// keeping the larger of two numbers.
func oneNumber() {
	totals := [replicas]int{}
	adds := [replicas]int{300, 120, 40}
	sum := 0

	for r, n := range adds {
		totals[r] += n
		sum += n
		fmt.Printf("  %-5s records %d GPU-seconds\n", names[r], n)
	}

	largest := max(totals[shelf], totals[desk], totals[bench])
	fmt.Printf("  after synchronising, every replica says %d; the jobs used %d\n", largest, sum)
}

Listing 9.6 Keeping the larger of two totals, in Rust. code/rust/counter/src/main.rs

/// Counts with a single number per replica and synchronises by keeping the
/// larger of two numbers.
fn one_number() {
    let adds = [300, 120, 40];

    for (name, n) in NAMES.iter().zip(adds) {
        println!("  {name:<5} records {n} GPU-seconds");
    }

    let largest = adds.iter().max().expect("three replicas");
    let sum: i64 = adds.iter().sum();
    println!("  after synchronising, every replica says {largest}; the jobs used {sum}");
}

Through a partition

The second experiment gives each replica a GCounter. The bench is cut off, the shelf and the desk merge with each other, and reports keep arriving on both sides. After the partition the rows are merged in an order chosen to be untidy, with one merge repeated.

Listing 9.7 Grow-only counters, a partition, and an untidy recovery, in Go. code/go/counter/main.go

// growOnly counts the same usage with a GCounter at each replica, through a
// partition that cuts the bench off from the other two.
func growOnly() {
	var at [replicas]GCounter

	for r, n := range [replicas]int{300, 120, 40} {
		at[r].Add(r, n)
	}

	fmt.Println("  the bench is cut off; shelf and desk exchange counters")
	at[shelf].Merge(at[desk])
	at[desk].Merge(at[shelf])
	at[shelf].Add(shelf, 60)
	at[bench].Add(bench, 25)

	for r := range at {
		show(r, at[r])
	}

	fmt.Println("  the network heals; counters are exchanged in no particular order, some twice")
	at[desk].Merge(at[bench])
	at[shelf].Merge(at[desk])
	at[bench].Merge(at[shelf])
	at[desk].Merge(at[shelf])
	at[desk].Merge(at[shelf])

	for r := range at {
		show(r, at[r])
	}
}

Listing 9.8 Grow-only counters, a partition, and an untidy recovery, in Rust. code/rust/counter/src/main.rs

/// Counts the same usage with a `GCounter` at each replica, through a
/// partition that cuts the bench off from the other two.
fn grow_only() {
    let mut at = [GCounter::default(); REPLICAS];

    for (r, n) in [300, 120, 40].into_iter().enumerate() {
        at[r].add(r, n);
    }

    println!("  the bench is cut off; shelf and desk exchange counters");
    at[SHELF].merge(at[DESK]);
    at[DESK].merge(at[SHELF]);
    at[SHELF].add(SHELF, 60);
    at[BENCH].add(BENCH, 25);

    for (r, c) in at.into_iter().enumerate() {
        show(r, c);
    }

    println!("  the network heals; counters are exchanged in no particular order, some twice");
    at[DESK].merge(at[BENCH]);
    at[SHELF].merge(at[DESK]);
    at[BENCH].merge(at[SHELF]);
    at[DESK].merge(at[SHELF]);
    at[DESK].merge(at[SHELF]);

    for (r, c) in at.into_iter().enumerate() {
        show(r, c);
    }
}

GPUs in use

The third experiment counts jobs starting and ending. One job’s end is reported to a different replica from the one that recorded its start, which is ordinary: a job’s machine and the replica that hears of its end need not be in the same place.

Listing 9.9 Starts and ends recorded at different replicas, in Go. code/go/counter/main.go

// upAndDown counts GPUs in use. A job's start and its end may be recorded at
// different replicas.
func upAndDown() {
	var at [replicas]PNCounter
	at[shelf].Add(shelf, 3)  // three jobs start on the Thors
	at[desk].Add(desk, 2)    // two start on the Sparks
	at[shelf].Add(shelf, -1) // one Thor job ends
	at[bench].Add(bench, -1) // the bench replica hears that a Spark job ended

	for r := range at {
		for other := range at {
			at[r].Merge(at[other])
		}
	}

	c := at[bench]
	fmt.Printf("  started %v, ended %v: %d GPUs in use\n", [replicas]int(c.Up), [replicas]int(c.Down), c.Value())
}

Listing 9.10 Starts and ends recorded at different replicas, in Rust. code/rust/counter/src/main.rs

/// Counts GPUs in use. A job's start and its end may be recorded at different
/// replicas.
fn up_and_down() {
    let mut at = [PnCounter::default(); REPLICAS];
    at[SHELF].add(SHELF, 3); // three jobs start on the Thors
    at[DESK].add(DESK, 2); // two start on the Sparks
    at[SHELF].add(SHELF, -1); // one Thor job ends
    at[BENCH].add(BENCH, -1); // the bench replica hears that a Spark job ended

    for r in 0..REPLICAS {
        for other in 0..REPLICAS {
            let theirs = at[other];
            at[r].merge(theirs);
        }
    }

    let c = at[BENCH];
    println!(
        "  started {}, ended {}: {} GPUs in use",
        row(c.up.0),
        row(c.down.0),
        c.value()
    );
}

A quota, two ways

The last two experiments enforce a quota of six GPUs. sharedLimit lets each replica check its own merged count. allowances gives each replica a share.

Listing 9.11 A limit checked against a local count, and a limit divided into shares, in Go. code/go/counter/main.go

// quota is the number of GPUs one team may use at once.
const quota = 6

// sharedLimit lets each replica admit a job whenever its own view of the
// count leaves room under the quota.
func sharedLimit() {
	var at [replicas]PNCounter
	at[shelf].Add(shelf, 3)
	at[desk].Add(desk, 1)
	at[shelf].Merge(at[desk])
	at[desk].Merge(at[shelf])
	fmt.Printf("  both replicas see %d of %d in use; then they lose contact\n", at[shelf].Value(), quota)

	for _, r := range []int{shelf, desk} {
		if at[r].Value()+2 <= quota {
			at[r].Add(r, 2)
			fmt.Printf("  %-5s admits a 2-GPU job: by its count, %d of %d are in use\n", names[r], at[r].Value(), quota)
		}
	}

	at[shelf].Merge(at[desk])
	fmt.Printf("  after they merge: %d of %d in use\n", at[shelf].Value(), quota)
}

// allowances gives each replica a share of the quota. A replica admits a job
// only from its own share, and shares move between replicas when they can
// reach each other.
func allowances() {
	share := [replicas]int{4, 2, 0}
	used := [replicas]int{3, 1, 0}
	fmt.Printf("  shares %v, in use %v; then the replicas lose contact\n", share, used)
	admit := func(r, gpus int) {
		if left := share[r] - used[r]; left < gpus {
			fmt.Printf("  %-5s refuses a %d-GPU job: its share has %d left\n", names[r], gpus, left)

			return
		}

		used[r] += gpus
		fmt.Printf("  %-5s admits a %d-GPU job\n", names[r], gpus)
	}
	admit(shelf, 2)
	admit(desk, 2)
	fmt.Println("  contact returns; desk gives its unused GPU to shelf")
	share[desk]--
	share[shelf]++
	admit(shelf, 2)
	fmt.Printf("  shares %v, in use %v: %d of %d\n", share, used, used[shelf]+used[desk]+used[bench], quota)
}

Listing 9.12 A limit checked against a local count, and a limit divided into shares, in Rust. code/rust/counter/src/main.rs

/// The number of GPUs one team may use at once.
const QUOTA: i64 = 6;

/// Lets each replica admit a job whenever its own view of the count leaves
/// room under the quota.
fn shared_limit() {
    let mut at = [PnCounter::default(); REPLICAS];
    at[SHELF].add(SHELF, 3);
    at[DESK].add(DESK, 1);
    let (shelf, desk) = (at[SHELF], at[DESK]);
    at[SHELF].merge(desk);
    at[DESK].merge(shelf);
    println!(
        "  both replicas see {} of {QUOTA} in use; then they lose contact",
        at[SHELF].value()
    );

    for r in [SHELF, DESK] {
        if at[r].value() + 2 <= QUOTA {
            at[r].add(r, 2);
            println!(
                "  {:<5} admits a 2-GPU job: by its count, {} of {QUOTA} are in use",
                NAMES[r],
                at[r].value()
            );
        }
    }

    let desk = at[DESK];
    at[SHELF].merge(desk);
    println!(
        "  after they merge: {} of {QUOTA} in use",
        at[SHELF].value()
    );
}

/// Gives each replica a share of the quota. A replica admits a job only from
/// its own share, and shares move between replicas when they can reach each
/// other.
fn allowances() {
    let mut share = [4, 2, 0];
    let mut used = [3, 1, 0];
    println!(
        "  shares {}, in use {}; then the replicas lose contact",
        row(share),
        row(used)
    );
    let admit = |share: &[i64; REPLICAS], used: &mut [i64; REPLICAS], r: usize, gpus: i64| {
        let left = share[r] - used[r];

        if left < gpus {
            println!(
                "  {:<5} refuses a {gpus}-GPU job: its share has {left} left",
                NAMES[r]
            );

            return;
        }

        used[r] += gpus;
        println!("  {:<5} admits a {gpus}-GPU job", NAMES[r]);
    };
    admit(&share, &mut used, SHELF, 2);
    admit(&share, &mut used, DESK, 2);
    println!("  contact returns; desk gives its unused GPU to shelf");
    share[DESK] -= 1;
    share[SHELF] += 1;
    admit(&share, &mut used, SHELF, 2);
    println!(
        "  shares {}, in use {}: {} of {QUOTA}",
        row(share),
        row(used),
        used.iter().sum::<i64>()
    );
}

The run

$ go run ./counter
one number per replica:
  shelf records 300 GPU-seconds
  desk  records 120 GPU-seconds
  bench records 40 GPU-seconds
  after synchronising, every replica says 300; the jobs used 460
one entry per replica, merged by taking the larger:
  the bench is cut off; shelf and desk exchange counters
  shelf holds [360 120 0] = 480
  desk  holds [300 120 0] = 420
  bench holds [0 0 65] = 65
  the network heals; counters are exchanged in no particular order, some twice
  shelf holds [360 120 65] = 545
  desk  holds [360 120 65] = 545
  bench holds [360 120 65] = 545
a count that goes down as well:
  started [3 2 0], ended [1 0 1]: 3 GPUs in use
a quota checked against each replica's own count:
  both replicas see 4 of 6 in use; then they lose contact
  shelf admits a 2-GPU job: by its count, 6 of 6 are in use
  desk  admits a 2-GPU job: by its count, 6 of 6 are in use
  after they merge: 8 of 6 in use
a quota divided into shares:
  shares [4 2 0], in use [3 1 0]; then the replicas lose contact
  shelf refuses a 2-GPU job: its share has 1 left
  desk  refuses a 2-GPU job: its share has 1 left
  contact returns; desk gives its unused GPU to shelf
  shelf admits a 2-GPU job
  shares [5 1 0], in use [5 1 0]: 6 of 6

With one number per replica, every replica says 300 after synchronising, and the jobs used 460.

With one entry per replica, the two sides of the partition disagree while it lasts: the shelf counts 480, the desk 420, and the bench 65. That is expected, since each is missing reports the others hold. After the untidy recovery all three hold the row [360 120 65] and the count 545, which is every report added once: 300, 120, 40, 60, and 25.

Animation 9.1 plays the second experiment, one merge at a time.

The count of GPUs in use comes to 3: five jobs started, and two ended, one of them recorded at the bench.

In the fourth experiment both replicas admit a job at “6 of 6”, and the merged count is 8 of 6. In the fifth, with shares of 4 and 2 and usage of 3 and 1, both replicas refuse the two-GPU job while apart. When contact returns, the desk gives its unused GPU to the shelf, the shelf admits the job, and the total in use is exactly 6. Figure 9.2 shows the shares before and after.

Where Kubernetes stands

Inside one cluster, Kubernetes takes the first design. A ResourceQuota object limits what a namespace may use, and the API server checks it when a pod is created. The check is exact because the API server is the one authority and etcd puts every write in a single order: the resource version of chapter 1 is that order made visible. A cluster whose control plane is cut off from part of itself stops accepting changes from that part, which is the consistent choice.

The techniques of this chapter apply above that level: usage that is gathered from many clusters into one bill, a team’s quota that spans clusters in several regions, a rate limit enforced by many copies of an API gateway. In each of these, no single API server sees every event, and the count has to be assembled from replicas that are sometimes apart.

The complete program

Listing 9.13 The complete program in Go. code/go/counter/main.go

// Command counter shows how replicas of a service that cannot always reach
// each other can keep one count.
//
// Three replicas of a usage service run beside the three groups of machines
// in our fleet: the shelf, the desk, and the bench. Each records the GPU time
// that jobs near it report. The program counts in four ways: with one number
// per replica, with a grow-only counter, with a counter that also goes down,
// and with a quota divided into allowances.
//
// Run it from code/go with:
//
//	go run ./counter
package main

import "fmt"

// The replicas, by index.
const (
	shelf = iota
	desk
	bench
	replicas
)

// names lists the replicas in index order.
var names = [replicas]string{"shelf", "desk", "bench"}

// GCounter is a count that only grows. It keeps one entry per replica, and
// each replica adds only to its own entry.
type GCounter [replicas]int

// Add records n more at the given replica.
func (c *GCounter) Add(replica, n int) { c[replica] += n }

// Merge takes, entry by entry, the larger of this counter's value and the
// other's. Merging is safe to repeat and to do in any order.
func (c *GCounter) Merge(other GCounter) {
	for i := range c {
		c[i] = max(c[i], other[i])
	}
}

// Value is the count: the sum of every replica's entry.
func (c GCounter) Value() int {
	total := 0

	for _, n := range c {
		total += n
	}

	return total
}


// PNCounter is a count that goes up and down, built from two counters that
// only grow: one for the increments and one for the decrements.
type PNCounter struct {
	Up, Down GCounter
}

// Add records a change of n, which may be negative, at the given replica.
func (c *PNCounter) Add(replica, n int) {
	if n >= 0 {
		c.Up.Add(replica, n)
	} else {
		c.Down.Add(replica, -n)
	}
}

// Merge merges both halves.
func (c *PNCounter) Merge(other PNCounter) {
	c.Up.Merge(other.Up)
	c.Down.Merge(other.Down)
}

// Value is the increments minus the decrements.
func (c PNCounter) Value() int { return c.Up.Value() - c.Down.Value() }


// oneNumber counts with a single number per replica and synchronises by
// keeping the larger of two numbers.
func oneNumber() {
	totals := [replicas]int{}
	adds := [replicas]int{300, 120, 40}
	sum := 0

	for r, n := range adds {
		totals[r] += n
		sum += n
		fmt.Printf("  %-5s records %d GPU-seconds\n", names[r], n)
	}

	largest := max(totals[shelf], totals[desk], totals[bench])
	fmt.Printf("  after synchronising, every replica says %d; the jobs used %d\n", largest, sum)
}


// show prints one replica's counter and its value.
func show(replica int, c GCounter) {
	fmt.Printf("  %-5s holds %v = %d\n", names[replica], [replicas]int(c), c.Value())
}

// growOnly counts the same usage with a GCounter at each replica, through a
// partition that cuts the bench off from the other two.
func growOnly() {
	var at [replicas]GCounter

	for r, n := range [replicas]int{300, 120, 40} {
		at[r].Add(r, n)
	}

	fmt.Println("  the bench is cut off; shelf and desk exchange counters")
	at[shelf].Merge(at[desk])
	at[desk].Merge(at[shelf])
	at[shelf].Add(shelf, 60)
	at[bench].Add(bench, 25)

	for r := range at {
		show(r, at[r])
	}

	fmt.Println("  the network heals; counters are exchanged in no particular order, some twice")
	at[desk].Merge(at[bench])
	at[shelf].Merge(at[desk])
	at[bench].Merge(at[shelf])
	at[desk].Merge(at[shelf])
	at[desk].Merge(at[shelf])

	for r := range at {
		show(r, at[r])
	}
}


// upAndDown counts GPUs in use. A job's start and its end may be recorded at
// different replicas.
func upAndDown() {
	var at [replicas]PNCounter
	at[shelf].Add(shelf, 3)  // three jobs start on the Thors
	at[desk].Add(desk, 2)    // two start on the Sparks
	at[shelf].Add(shelf, -1) // one Thor job ends
	at[bench].Add(bench, -1) // the bench replica hears that a Spark job ended

	for r := range at {
		for other := range at {
			at[r].Merge(at[other])
		}
	}

	c := at[bench]
	fmt.Printf("  started %v, ended %v: %d GPUs in use\n", [replicas]int(c.Up), [replicas]int(c.Down), c.Value())
}


// quota is the number of GPUs one team may use at once.
const quota = 6

// sharedLimit lets each replica admit a job whenever its own view of the
// count leaves room under the quota.
func sharedLimit() {
	var at [replicas]PNCounter
	at[shelf].Add(shelf, 3)
	at[desk].Add(desk, 1)
	at[shelf].Merge(at[desk])
	at[desk].Merge(at[shelf])
	fmt.Printf("  both replicas see %d of %d in use; then they lose contact\n", at[shelf].Value(), quota)

	for _, r := range []int{shelf, desk} {
		if at[r].Value()+2 <= quota {
			at[r].Add(r, 2)
			fmt.Printf("  %-5s admits a 2-GPU job: by its count, %d of %d are in use\n", names[r], at[r].Value(), quota)
		}
	}

	at[shelf].Merge(at[desk])
	fmt.Printf("  after they merge: %d of %d in use\n", at[shelf].Value(), quota)
}

// allowances gives each replica a share of the quota. A replica admits a job
// only from its own share, and shares move between replicas when they can
// reach each other.
func allowances() {
	share := [replicas]int{4, 2, 0}
	used := [replicas]int{3, 1, 0}
	fmt.Printf("  shares %v, in use %v; then the replicas lose contact\n", share, used)
	admit := func(r, gpus int) {
		if left := share[r] - used[r]; left < gpus {
			fmt.Printf("  %-5s refuses a %d-GPU job: its share has %d left\n", names[r], gpus, left)

			return
		}

		used[r] += gpus
		fmt.Printf("  %-5s admits a %d-GPU job\n", names[r], gpus)
	}
	admit(shelf, 2)
	admit(desk, 2)
	fmt.Println("  contact returns; desk gives its unused GPU to shelf")
	share[desk]--
	share[shelf]++
	admit(shelf, 2)
	fmt.Printf("  shares %v, in use %v: %d of %d\n", share, used, used[shelf]+used[desk]+used[bench], quota)
}


func main() {
	fmt.Println("one number per replica:")
	oneNumber()
	fmt.Println("one entry per replica, merged by taking the larger:")
	growOnly()
	fmt.Println("a count that goes down as well:")
	upAndDown()
	fmt.Println("a quota checked against each replica's own count:")
	sharedLimit()
	fmt.Println("a quota divided into shares:")
	allowances()
}

Listing 9.14 The complete program in Rust. code/rust/counter/src/main.rs

//! Shows how replicas of a service that cannot always reach each other can
//! keep one count.
//!
//! Three replicas of a usage service run beside the three groups of machines
//! in our fleet: the shelf, the desk, and the bench. Each records the GPU time
//! that jobs near it report. The program counts in four ways: with one number
//! per replica, with a grow-only counter, with a counter that also goes down,
//! and with a quota divided into allowances.
//!
//! Run it from code/rust with:
//!
//! ```text
//! cargo run -p counter
//! ```

/// The number of replicas.
const REPLICAS: usize = 3;
/// The replicas, by index.
const SHELF: usize = 0;
const DESK: usize = 1;
const BENCH: usize = 2;
/// The replicas' names, in index order.
const NAMES: [&str; REPLICAS] = ["shelf", "desk", "bench"];

/// Formats a row of numbers as Go prints an array: `[1 2 3]`.
fn row(values: [i64; REPLICAS]) -> String {
    format!("[{}]", values.map(|v| v.to_string()).join(" "))
}

/// A count that only grows. It keeps one entry per replica, and each replica
/// adds only to its own entry.
#[derive(Clone, Copy, Default)]
struct GCounter([i64; REPLICAS]);

impl GCounter {
    /// Records n more at the given replica.
    fn add(&mut self, replica: usize, n: i64) {
        self.0[replica] += n;
    }

    /// Takes, entry by entry, the larger of this counter's value and the
    /// other's. Merging is safe to repeat and to do in any order.
    fn merge(&mut self, other: GCounter) {
        for (mine, theirs) in self.0.iter_mut().zip(other.0) {
            *mine = (*mine).max(theirs);
        }
    }

    /// The count: the sum of every replica's entry.
    fn value(self) -> i64 {
        self.0.iter().sum()
    }
}

/// A count that goes up and down, built from two counters that only grow: one
/// for the increments and one for the decrements.
#[derive(Clone, Copy, Default)]
struct PnCounter {
    up: GCounter,
    down: GCounter,
}

impl PnCounter {
    /// Records a change of n, which may be negative, at the given replica.
    fn add(&mut self, replica: usize, n: i64) {
        if n >= 0 {
            self.up.add(replica, n);
        } else {
            self.down.add(replica, -n);
        }
    }

    /// Merges both halves.
    fn merge(&mut self, other: PnCounter) {
        self.up.merge(other.up);
        self.down.merge(other.down);
    }

    /// The increments minus the decrements.
    fn value(self) -> i64 {
        self.up.value() - self.down.value()
    }
}

/// Counts with a single number per replica and synchronises by keeping the
/// larger of two numbers.
fn one_number() {
    let adds = [300, 120, 40];

    for (name, n) in NAMES.iter().zip(adds) {
        println!("  {name:<5} records {n} GPU-seconds");
    }

    let largest = adds.iter().max().expect("three replicas");
    let sum: i64 = adds.iter().sum();
    println!("  after synchronising, every replica says {largest}; the jobs used {sum}");
}

/// Prints one replica's counter and its value.
fn show(replica: usize, c: GCounter) {
    println!("  {:<5} holds {} = {}", NAMES[replica], row(c.0), c.value());
}

/// Counts the same usage with a `GCounter` at each replica, through a
/// partition that cuts the bench off from the other two.
fn grow_only() {
    let mut at = [GCounter::default(); REPLICAS];

    for (r, n) in [300, 120, 40].into_iter().enumerate() {
        at[r].add(r, n);
    }

    println!("  the bench is cut off; shelf and desk exchange counters");
    at[SHELF].merge(at[DESK]);
    at[DESK].merge(at[SHELF]);
    at[SHELF].add(SHELF, 60);
    at[BENCH].add(BENCH, 25);

    for (r, c) in at.into_iter().enumerate() {
        show(r, c);
    }

    println!("  the network heals; counters are exchanged in no particular order, some twice");
    at[DESK].merge(at[BENCH]);
    at[SHELF].merge(at[DESK]);
    at[BENCH].merge(at[SHELF]);
    at[DESK].merge(at[SHELF]);
    at[DESK].merge(at[SHELF]);

    for (r, c) in at.into_iter().enumerate() {
        show(r, c);
    }
}

/// Counts GPUs in use. A job's start and its end may be recorded at different
/// replicas.
fn up_and_down() {
    let mut at = [PnCounter::default(); REPLICAS];
    at[SHELF].add(SHELF, 3); // three jobs start on the Thors
    at[DESK].add(DESK, 2); // two start on the Sparks
    at[SHELF].add(SHELF, -1); // one Thor job ends
    at[BENCH].add(BENCH, -1); // the bench replica hears that a Spark job ended

    for r in 0..REPLICAS {
        for other in 0..REPLICAS {
            let theirs = at[other];
            at[r].merge(theirs);
        }
    }

    let c = at[BENCH];
    println!(
        "  started {}, ended {}: {} GPUs in use",
        row(c.up.0),
        row(c.down.0),
        c.value()
    );
}

/// The number of GPUs one team may use at once.
const QUOTA: i64 = 6;

/// Lets each replica admit a job whenever its own view of the count leaves
/// room under the quota.
fn shared_limit() {
    let mut at = [PnCounter::default(); REPLICAS];
    at[SHELF].add(SHELF, 3);
    at[DESK].add(DESK, 1);
    let (shelf, desk) = (at[SHELF], at[DESK]);
    at[SHELF].merge(desk);
    at[DESK].merge(shelf);
    println!(
        "  both replicas see {} of {QUOTA} in use; then they lose contact",
        at[SHELF].value()
    );

    for r in [SHELF, DESK] {
        if at[r].value() + 2 <= QUOTA {
            at[r].add(r, 2);
            println!(
                "  {:<5} admits a 2-GPU job: by its count, {} of {QUOTA} are in use",
                NAMES[r],
                at[r].value()
            );
        }
    }

    let desk = at[DESK];
    at[SHELF].merge(desk);
    println!(
        "  after they merge: {} of {QUOTA} in use",
        at[SHELF].value()
    );
}

/// Gives each replica a share of the quota. A replica admits a job only from
/// its own share, and shares move between replicas when they can reach each
/// other.
fn allowances() {
    let mut share = [4, 2, 0];
    let mut used = [3, 1, 0];
    println!(
        "  shares {}, in use {}; then the replicas lose contact",
        row(share),
        row(used)
    );
    let admit = |share: &[i64; REPLICAS], used: &mut [i64; REPLICAS], r: usize, gpus: i64| {
        let left = share[r] - used[r];

        if left < gpus {
            println!(
                "  {:<5} refuses a {gpus}-GPU job: its share has {left} left",
                NAMES[r]
            );

            return;
        }

        used[r] += gpus;
        println!("  {:<5} admits a {gpus}-GPU job", NAMES[r]);
    };
    admit(&share, &mut used, SHELF, 2);
    admit(&share, &mut used, DESK, 2);
    println!("  contact returns; desk gives its unused GPU to shelf");
    share[DESK] -= 1;
    share[SHELF] += 1;
    admit(&share, &mut used, SHELF, 2);
    println!(
        "  shares {}, in use {}: {} of {QUOTA}",
        row(share),
        row(used),
        used.iter().sum::<i64>()
    );
}

fn main() {
    println!("one number per replica:");
    one_number();
    println!("one entry per replica, merged by taking the larger:");
    grow_only();
    println!("a count that goes down as well:");
    up_and_down();
    println!("a quota checked against each replica's own count:");
    shared_limit();
    println!("a quota divided into shares:");
    allowances();
}

References

• M. Shapiro, N. Preguiça, C. Baquero, M. Zawirski, “A comprehensive study of Convergent and Commutative Replicated Data Types”, INRIA research report 7506, 2011.
• V. Balegas and others, “Extending Eventually Consistent Cloud Databases for Enforcing Numeric Invariants”, 34th IEEE Symposium on Reliable Distributed Systems (SRDS), 2015, which describes the bounded counter with transferable shares.
• S. Gilbert and N. Lynch, “Brewer’s conjecture and the feasibility of consistent, available, partition-tolerant web services”, ACM SIGACT News, 33(2), 2002.
• Kubernetes, Resource quotas: https://kubernetes.io/docs/concepts/policy/resource-quotas/
• etcd, KV API guarantees: https://etcd.io/docs/latest/learning/api_guarantees/