Samuele Resca — A practical approach to read-write quorum systems

ℹ️ The 2nd part of this post is available: [Part 2] A practical approach to read-write quorum systems.

Quorum systems allow consistency of replicated data; every time a group of servers needs to agree on something, a quorum is involved in the decisions.

For example, leaderless databases such as Dynamo (not to be confused with DynamoDB) use a consistency protocol based on quorums¹. Furthermore, most of the consensus algorithms such as Zab, Raft, Paxos are based on quorums².

Read-write quorums define two configurable parameters, R and W. R is the minimum number of nodes that must participate in a read operation, and W the minimum number of nodes that must participate in a write operation.

A few weeks ago, I came across this tweet:

Majority quorums are pervasive in strongly consistent distributed systems yet there are many alternatives with better scalability and performance. Quoracle is a new open-source tool to find an optimal quorum system for a given distributed architecture. https://t.co/d4UpFNdQuf pic.twitter.com/V2SaXYmkoP
— Heidi Howard (@heidiann360) April 13, 2021

The tweet refers to the following publication:

Read-Write Quorum Systems Made Practical - Michael Whittaker, Aleksey Charapko, Joseph M. Hellerstein, Heidi Howard, Ion Stoica

The paper reviews some concepts of the quorum systems, and it presents a concrete tool named "Quoracle" that explores the trade-offs of the read-write quorum systems.

Quoracle provides an alternative to the majority quorum systems that are widely adopted in distributed systems. The majority quorum can be defined as

$$ \frac{n}{2} + 1 \space where \space n = n. \space of \space nodes $$

In the case of a read-write quorum system, the majority is represented in a similar way:

$$ r = w = \frac{n}{2} + 1 $$

where $r$ and $w$ are the read and write quorums.

Below are some notes I took while I was reading the paper and a detailed look at the proposed implementation of Quoracle. The topics discussed are listed below:

Quorum system definitions
Quoracle overview
Optimal strategy problem implementation
Strategy representation
Handling quorum failures
Optimal strategy search

Quorum system definitions

The paper summarises the definitions and the characteristics of the read-write quorum systems. These concepts are helpful to understand the theory of quorum systems and the implementation of Quoracle.

Given a list of nodes represented as follows:

$$ X = {{x_1,..,x_n}} $$

A read-write quorum system is defined as $Q = (R, W)$ where $R$ represents the read quorums, and $W$ represents the write quorums. $R$ and $W$ are sets of subsets of the list of nodes defined in $X$.

An additional constraint asserts that every read and write quorum ($r$ and $w$) must intersect. This is represented with the following definition:

$$ \forall r \in R, \forall w \in W : r \cap w \neq \emptyset $$

It is possible to illustrate a quorum system with a grid as follows:

2 by 3 grid quorum system Q2×3

The grid above describes a $ Q_{2x3} $ quorum system defined as $ Q_{2x3} = (abc + def, ad + be + cf)$.

Note that multiplication represents the nodes that are part of the same set while the addition separates the groups. This approach will be helpful to define read and write sets in Quoracle.

Fault tolerance

The paper describes the read fault tolerance of a quorum system as the largest number, called $f$, of nodes that can fail such that a read quorum is still available. The same definition is applied to the write fault tolerance.

The fault tolerance of a quorum system is the minimum between the read fault tolerance and the write fault tolerance. For example, the $Q_{2x3}$ quorum system defined above has a read fault tolerance of 1 because if one node is down, it means that only a single read quorum will be available. On the other hand, the write fault tolerance is 2 because if two nodes fail, then there will be only 1 write quorum available.

In this case, the fault tolerance of the quorum system is equal to 1.

Strategy

The paper describes the concept of strategy of a read-write quorum system: the strategy of a quorum system, represented in the paper as $\sigma$, decides which quorum to contact.

The strategy of a read-write quorum system is represented as:

$$ \sigma = (\sigma_R, \sigma_W) $$

where $\sigma_R$ and $\sigma_W$ are probability distributions: $\sigma_R(r)$ and $\sigma_W(w)$ are respectively the probabilities that the strategy chooses a read quorum $r$ and a write quorum $w$.

Load, capacity and the read fraction

The read load of a node $x$, represented as $load_{\sigma_R}(x)$ is the probability that $\sigma_R$ chooses a read quorum $r$ that contains the node $x$. The same approach can be taken for the $load_{\sigma_W}(x)$.

For calculating the total load of a node $x$, we need to introduce another parameter: the read fraction.

The read fraction, represented as $f_r$, is the percentage of reads vs. write operations in a workload. For example, a workload with 50% read operations and 50% write operations has a read fraction $ f_r = 0.5 $.

Therefore, the total load of x can be represented by the following definition: $$ load(x) = f_rload_{\sigma_R}(x) + (1 - f_r)load_{\sigma_W}(x) $$

The load of the quorum system is the load of the optimal strategy, defined as the strategy that brings the lowest load.

The capacity can be represented as the inverse of the load:

$$ C = \frac{1}{L} \hspace{0.5cm} where \hspace{0.5cm} L = load $$

The higher the load on the quorums, the lower the capacity of the quorum system.

In conclusion, these are some of the definitions illustrated in the paper. This section gives a good overview of the possible parameters that can influence the performance and the efficiency of read-write quorum systems. The following section is more practical, and it shows the application of these definitions on the implementation of Quoracle.

Quoracle overview

The concepts introduced in the previous section help understand the implementation of the quoracle library described in the paper, available on GitHub: mwhittaker/quoracle.

Note that the quoracle lib is distributed on PyPI, and it can be referred to in a project using the following command:

pip install quoracle

The paper provides an initial example similar to this:

from quoracle import QuorumSystem, Node

if __name__ == '__main__':
    a, b, c = Node('a'), Node('b'), Node('c')
    majority = QuorumSystem(reads=a*b + b*c + a*c)
    print(majority.load(read_fraction=1))     # 2/3
    print(majority.capacity(read_fraction=1)) # 3/2

The snippet above declares nodes using the Node class and the corresponding quorum system defined by the QuorumSystem class. The Node class is used to describe the nodes that are part of the $X$ set:

"""
Original code available at: https://github.com/mwhittaker/quoracle
"""
class Node(Expr[T]):
    def __init__(self,
                 x: T,
                 capacity: Optional[float] = None,
                 read_capacity: Optional[float] = None,
                 write_capacity: Optional[float] = None,
                 latency: datetime.timedelta = None) -> None:
        self.x = x

        # A user either specifies capacity or (read_capacity and
        # write_capacity), but not both.
        if (capacity is None and
                read_capacity is None and
                write_capacity is None):
            self.read_capacity = 1.0
            self.write_capacity = 1.0
        elif (capacity is not None and
              read_capacity is None and
              write_capacity is None):
            self.read_capacity = capacity
            self.write_capacity = capacity
        elif (capacity is None and
              read_capacity is not None and
              write_capacity is not None):
            self.read_capacity = read_capacity
            self.write_capacity = write_capacity
        else:
            raise ValueError('You must specify capacity or (read_capacity '
                             'and write_capacity)')

        if latency is None:
            self.latency = datetime.timedelta(seconds=1)
        else:
            self.latency = latency

Each Node instance has a name, a read_capacity, a write_capacity, and a latency attribute.

The read_capacity, write_capacity and latency attributes represent heterogeneous nodes.

When a read/write capacity is specified for a node, it is used to normalise the node load during the linear optimisation; we will go through the detailed process later in this post.

By default, the latency of a Node is 1s. The attribute is used in the linear optimisation process for calculating the latency of the quorum systems.

The multiplication (*) and sum (+) operations are defined by the Expr class that is extended by the Node class:

"""
Original code available at: https://github.com/mwhittaker/quoracle
"""
class Expr(Generic[T]):
    def __add__(self, rhs: 'Expr[T]') -> 'Expr[T]':
        def _or(lhs: Expr[T], rhs: Expr[T]) -> 'Or[T]':
            if isinstance(lhs, Or) and isinstance(rhs, Or):
                return Or(lhs.es + rhs.es)
            elif isinstance(lhs, Or):
                return Or(lhs.es + [rhs])
            elif isinstance(rhs, Or):
                return Or([lhs] + rhs.es)
            else:
                return Or([lhs, rhs])

        return _or(self, rhs)

    def __mul__(self, rhs: 'Expr[T]') -> 'Expr[T]':
        def _and(lhs: Expr[T], rhs: Expr[T]) -> 'And[T]':
            if isinstance(lhs, And) and isinstance(rhs, And):
                return And(lhs.es + rhs.es)
            elif isinstance(lhs, And):
                return And(lhs.es + [rhs])
            elif isinstance(rhs, And):
                return And([lhs] + rhs.es)
            else:
                return And([lhs, rhs])

        return _and(self, rhs)

Each Node instance extends the Expr type; this approach allows composing expressions of nodes, such as the following quorum definition:

  a, b, c = Node('a'), Node('b'), Node('c')
  q = QuorumSystem(reads=a * b + b * c + a * c)

Another key component is the QuorumSystem class: it encapsulates the read and write quorums of the system, and it defines some methods for calculating the optimal strategy: the load, the latency, and the network load.

The snippet below defines the QuorumSystem class implemented by the library and the signature of the key methods exposed by the class:

"""
Original code available at: https://github.com/mwhittaker/quoracle
"""
class QuorumSystem(Generic[T]):
    def __init__(self, reads: Optional[Expr[T]] = None,
                       writes: Optional[Expr[T]] = None) -> None:
        if reads is not None and writes is not None:
            optimal_writes = reads.dual()
            if not all(optimal_writes.is_quorum(wq) for wq in writes.quorums()):
                raise ValueError(
                    'Not all read quorums intersect all write quorums')

            self.reads = reads
            self.writes = writes
        elif reads is not None and writes is None:
            self.reads = reads
            self.writes = reads.dual()
        elif reads is None and writes is not None:
            self.reads = writes.dual()
            self.writes = writes
        else:
            raise ValueError('A QuorumSystem must be instantiated with a set '
                             'of read quorums or a set of write quorums')

        self.x_to_node = {node.x: node for node in self.nodes()}
        
    def load(self,
             optimize: str = LOAD,
             load_limit: Optional[float] = None,
             network_limit: Optional[float] = None,
             latency_limit: Optional[datetime.timedelta] = None,
             read_fraction: Optional[Distribution] = None,
             write_fraction: Optional[Distribution] = None,
             f: int = 0) -> float:
         ...

    def capacity(self,
                 optimize: str = LOAD,
                 load_limit: Optional[float] = None,
                 network_limit: Optional[float] = None,
                 latency_limit: Optional[datetime.timedelta] = None,
                 read_fraction: Optional[Distribution] = None,
                 write_fraction: Optional[Distribution] = None,
                 f: int = 0) -> float:
        ...

    def network_load(self,
                     optimize: str = LOAD,
                     load_limit: Optional[float] = None,
                     network_limit: Optional[float] = None,
                     latency_limit: Optional[datetime.timedelta] = None,
                     read_fraction: Optional[Distribution] = None,
                     write_fraction: Optional[Distribution] = None,
                     f: int = 0) -> float:
        ...

    def latency(self,
                optimize: str = LOAD,
                load_limit: Optional[float] = None,
                network_limit: Optional[float] = None,
                latency_limit: Optional[datetime.timedelta] = None,
                read_fraction: Optional[Distribution] = None,
                write_fraction: Optional[Distribution] = None,
                f: int = 0) -> float:
        ...

The snippet above defines four methods for the class QuorumSystem: load, capacity, network_load, latency. We will have a detailed look at the implementation of the methods later in this section. First, however, all these methods have a similar signature: they require an optimize parameter representing the objective of the optimisation, either the read_fraction or the write_fraction parameters (already described in the theory section), and some additional constraints used by the optimisation of the quorum strategy, such as the load_limit, network_limit, latency_limit parameters.

As you may have noticed, the initialisation of the quorum system only provides the following read quorums:

ab + bc + ac

In this case, the library will derive the optimal write quorums by applying a dual operation to the given quorum. For example, the result of the dual process on the ab + bc + ac quorum calculates:

(a + b) * (b + c) * (a + c)

Once the QuorumSystem is initialised, it is possible to calculate the load, the latency, and the network_load of the system.

Optimal strategy problem implementation

This section digs into the optimal strategy implementation. The load, capacity, network_load and latency methods in the QuorumSystem class use the concept of optimal strategy, referred to in the paper as $ \sigma^{\ast} = (\sigma_R^{\ast}, \sigma_W^{\ast}) $.

A strategy can be optimal with respect to the load, the network, or the latency. The objective of the optimisation can be configured using the optimize parameter.

The internal implementation turns the optimisation into a linear programming problem, and it relies on PuLP to find the optimal strategy.

In the following sections, we cover the different optimisation objectives provided by Quoracle.

Load optimisation

In case optimize == LOAD, the objective of the linear optimisation is the load. The implementation of the load optimisation works as follows:

'''
Original code available at:  https://github.com/mwhittaker/quoracle/blob/main/quoracle/quorum_system.py#L541
'''

#...

def load(problem: pulp.LpProblem,
         read_fraction: Dict[float, float]) -> pulp.LpAffineExpression:
    return sum(p * fr_load(problem, fr)
               for (fr, p) in read_fraction.items())


def fr_load(problem: pulp.LpProblem, fr: float) -> pulp.LpAffineExpression:
    l = pulp.LpVariable(f'l_{fr}', 0, 1)

    for node in self.nodes():
        x = node.x
        x_load: pulp.LpAffineExpression = 0

        if x in x_to_read_quorum_vars:
            vs = x_to_read_quorum_vars[x]
            x_load += fr * sum(vs) / self.node(x).read_capacity

        if x in x_to_write_quorum_vars:
            vs = x_to_write_quorum_vars[x]
            x_load += (1 - fr) * sum(vs) / self.node(x).write_capacity

        problem += (x_load <= l, f'{x}{fr}')

    return l

# Form the linear program.
problem = pulp.LpProblem("optimal_strategy", pulp.LpMinimize)

The implementation minimises the load using the following definition:

$$ \frac{f_r}{cap_R(x)} \sum_{{r \in R | x \in r }} p_r + \frac{1 - f_r}{cap_W(x)} \sum_{{w \in W | x \in w }} p_w \leq L_{f_r} $$

For each read_fraction ($f_r$) in the QuorumSystem, the load method proceeds by computing the load by calling the fr_load method.

As a first step, the fr_load method defines the $L_{f_r}$ variable, representing the load for that read fraction. It continues by adding the read quorum part of the definition above:

x_load += fr * sum(vs) / self.node(x).read_capacity

the same for the write quorum part:

x_load += (1 - fr) * sum(vs) / self.node(x).write_capacity

As a final step, it proceeds by finalising the constraint using the instruction:

problem += (x_load <= l, f'{x}{fr}')

Network load optimisation

The following expression defines the network load.

$$ f_r ( \sum_{r \in R} p_r \cdot |r|) + (1 - f_r) ( \sum_{w \in W} p_w \cdot |w|) $$

Where $|r|$ and $|w|$ are the length of the read and write quorums sets.

In practice, the library implements the following code:

"""
Original code available at: https://github.com/mwhittaker/quoracle/blob/master/quoracle/quorum_system.py#L498
"""
def network() -> pulp.LpAffineExpression:
    reads = fr * sum(
        v * len(rq)
        for (rq, v) in zip(read_quorums, read_quorum_vars)
    )
    writes = (1 - fr) * sum(
        v * len(wq)
        for (wq, v) in zip(write_quorums, write_quorum_vars)
    )
    return reads + writes

If optimize == NETWORK, quoracle will use the network function above as a target objective for the optimal strategy. The implementation follows the network load definition mentioned above. Each read and write quorum multiplies the quorum length by the read_quorum_vars and write_quorum_vars, and it sums the values together.

If $f_r$ is a distribution, the network load is computed as a weighted average of all the network loads derived by every single value in the distribution.

Latency optimisation

The latency is another important aspect of a quorum system. In the real world, nodes are heterogeneous, and the latency depends on the network conditions and the status of the node that we are contacting.

The paper defines the latency as:

$$ f_r ( \sum_{r \in R} p_r \cdot latency(r)) + (1 - f_r) ( \sum_{w \in W} p_w \cdot latency(w)) $$

In practice, in case optimize == LATENCY, Quoracle uses the following approach for calculating the latency:

"""
Original code available at: https://github.com/mwhittaker/quoracle/blob/d83a811a905a51fc13ebc00ddac036d559463380/quoracle/quorum_system.py#L509
"""
def latency() -> pulp.LpAffineExpression:
    reads = fr * sum(
        v * self._read_quorum_latency(quorum).total_seconds()
        for (rq, v) in zip(read_quorums, read_quorum_vars)
        for quorum in [{self.node(x) for x in rq}]
    )
    writes = (1 - fr) * sum(
        v * self._write_quorum_latency(quorum).total_seconds()
        for (wq, v) in zip(write_quorums, write_quorum_vars)
        for quorum in [{self.node(x) for x in wq}]
    )
    return reads + writes

The Node class has a configurable latency. The implementation builds the latency minimisation problem by multiplying the latency in seconds assigned to each node with the read and write quorums. If $f_r$ represents a distribution, then the implementation uses the weighted average of the values in the distribution to get the final result.

Additional constraints

As we saw previously, the target of linear optimisation depends on the optimize parameter. Therefore, the resulting optimal strategy will be optimal regarding the load, the network load, or the latency.

In addition, it is possible to specify some additional constraints for the optimal strategy: these constraints are limits on the load, the network, or the latency.

Note that it is not possible to optimise for load and at the same time specify a load limit to the strategy; the same restriction is valid for the network and the latency.

The additional constraints use the same definitions of load, network load, and latency we saw previously. This example shows a final problem that defines an optimal load strategy with some constraints on the network and the latency.

Let's take for example the following optimisation query:

q = QuorumSystem(reads=a * b + b * c + a * c)

q.load(optimize=LOAD, read_fraction=1, network_limit=3, latency_limit=datetime.timedelta(seconds=2))

The QuorumSystem is defined by some quorums composed of three nodes: a, b, c. We want to get the optimal load (optimize==LOAD), with a read_fraction of 1 (100% reads workload), a network_limit==3 and a latency_limit of 2 seconds (by default, every Node has a latency of 1s if we do not specify anything). The resulting optimisation problem that finds the optimal strategy looks like this:

optimal_strategy:
    MINIMIZE
        1.0*l_1.0 + 0.0
    SUBJECT TO
        valid_read_strategy: r0 + r1 + r2 = 1
        valid_write_strategy: w0 + w1 + w2 + w3 + w4 + w5 + w6 + w7 = 1
        a1.0: - l_1.0 + r0 + r2 <= 0
        b1.0: - l_1.0 + r0 + r1 <= 0
        c1.0: - l_1.0 + r1 + r2 <= 0
        network_limit: 2 r0 + 2 r1 + 2 r2 <= 3
        latency_limit: r0 + r1 + r2 <= 2

As you can see, the optimal strategy problem tries to minimise the load with respect to some constraints:

valid read/write strategies: the sum of all the read/write probabilities is 1;
the a1.0, b1.0, c1.0 are the constraints which represent the load optimisation. In detail, the constraint a1.0 representing the load for node a, uses the constraint r0 + r2 <= l_1.0;
the network_limit must be less than three as we specified in the load method call above;
the latency_limit must be less than 2s as we specified in the load method call above;

The reference to the optimal_strategy problem is then optimised, and the resulting values, encapsulated in a pulp.LpVariable instance, are passed to a new Strategy type.

Strategy representation

Quoracle uses a Strategy class to represent the concept of strategy (σ) for the quorum system. The following snippet of code describes the implementation and the constructor of a Strategy[T] type:

"""
Original code available at: https://github.com/mwhittaker/quoracle/blob/main/quoracle/quorum_system.py#L596
"""
class Strategy(Generic[T]):
    def __init__(self,
                 qs: QuorumSystem[T],
                 sigma_r: Dict[FrozenSet[T], float],
                 sigma_w: Dict[FrozenSet[T], float]) -> None:
        self.qs = qs
        self.sigma_r = sigma_r
        self.sigma_w = sigma_w

        # The probability that x is chosen as part of a read quorum.
        self.x_read_probability: Dict[T, float] = collections.defaultdict(float)
        for (read_quorum, p) in self.sigma_r.items():
            for x in read_quorum:
                self.x_read_probability[x] += p

        # The probability that x is chosen as part of a write quorum.
        self.x_write_probability: Dict[T, float] = collections.defaultdict(float)
        for (write_quorum, weight) in self.sigma_w.items():
            for x in write_quorum:
                self.x_write_probability[x] += weight

The class extends a Generic[T] type, and it wraps a reference to a QuorumSystem[T] type, a sigma_r, and a sigma_w attribute; the last two properties represent the definition of strategy we saw in the first section:

$$ \sigma = (\sigma_R, \sigma_W) $$

The Strategy constructor initialises two dictionaries: the x_read_probability and x_write_probability. They provide a map between node x and its probability of getting selected as part of a quorum.

The Strategy class exposes the methods for retrieving the read and write quorums and the nodes that are part of the quorum systems:

"""
Original code available at: https://github.com/mwhittaker/quoracle/blob/master/quoracle/quorum_system.py#L622
"""
class Strategy(Generic[T]):
    
    # ...

    def quorum_system(self) -> QuorumSystem[T]:
        return self.qs

    def node(self, x: T) -> Node[T]:
        return self.qs.node(x)

    def nodes(self) -> Set[Node[T]]:
        return self.qs.nodes()

    def get_read_quorum(self) -> Set[T]:
        return set(random.choices(list(self.sigma_r.keys()),
                                  weights=list(self.sigma_r.values()))[0])

    def get_write_quorum(self) -> Set[T]:
        return set(random.choices(list(self.sigma_w.keys()),
                                  weights=list(self.sigma_w.values()))[0])

The get_read_quorum and the get_write_quorum methods pick the read/write set depending on the probability of getting selected.

Finally, the Strategy class implements the methods for calculating key characteristics we discussed previously, such as the load, the capacity, the network_load, and the latency:

"""
Original code available at: https://github.com/mwhittaker/quoracle/blob/master/quoracle/quorum_system.py#L639
"""
class Strategy(Generic[T]):
  
    # ...

    def load(self,
             read_fraction: Optional[Distribution] = None,
             write_fraction: Optional[Distribution] = None) -> float:
        d = distribution.canonicalize_rw(read_fraction, write_fraction)
        return sum(p * self._load(fr) for (fr, p) in d.items())

    def capacity(self,
                 read_fraction: Optional[Distribution] = None,
                 write_fraction: Optional[Distribution] = None) -> float:
        d = distribution.canonicalize_rw(read_fraction, write_fraction)
        return sum(p * 1 / self._load(fr) for (fr, p) in d.items())

    def network_load(self,
                     read_fraction: Optional[Distribution] = None,
                     write_fraction: Optional[Distribution] = None) -> float:
        d = distribution.canonicalize_rw(read_fraction, write_fraction)
        fr = sum(p * fr for (fr, p) in d.items())
        reads = fr * sum(p * len(rq) for (rq, p) in self.sigma_r.items())
        writes = (1 - fr) * sum(p * len(wq) for (wq, p) in self.sigma_w.items())
        return reads + writes

    # mypy doesn't like calling sum with timedeltas.
    @no_type_check
    def latency(self,
                read_fraction: Optional[Distribution] = None,
                write_fraction: Optional[Distribution] = None) \
                -> datetime.timedelta:
        d = distribution.canonicalize_rw(read_fraction, write_fraction)
        fr = sum(p * fr for (fr, p) in d.items())

        reads = fr * sum((
            p * self.qs._read_quorum_latency({self.node(x) for x in rq})
            for (rq, p) in self.sigma_r.items()
        ), datetime.timedelta(seconds=0))

        writes = (1 - fr) * sum((
            p * self.qs._write_quorum_latency({self.node(x) for x in wq})
            for (wq, p) in self.sigma_w.items()
        ), datetime.timedelta(seconds=0))

        return reads + writes

    def node_load(self,
                  node: Node[T],
                  read_fraction: Optional[Distribution] = None,
                  write_fraction: Optional[Distribution] = None) -> float:
        d = distribution.canonicalize_rw(read_fraction, write_fraction)
        return sum(p * self._node_load(node, fr) for (fr, p) in d.items())

    def _load(self, fr: float) -> float:
        return max(self._node_load(node, fr) for node in self.nodes())

    def _node_load(self, node: Node[T], fr: float) -> float:
        fw = 1 - fr
        return (fr * self.x_read_probability[node.x] / node.read_capacity +
                fw * self.x_write_probability[node.x] / node.write_capacity)

The implementation works similarly to the constraints definitions that define the optimal strategy problem we discussed in the previous sections. Again, it is possible to pass the read_fraction and the write_fraction distributions to represent heterogeneous workloads.

Handling quorum failures

In distributed systems, failures are frequent. The two main approaches adopted in quorum systems are:

contact every node in the quorum system in case some of them are failing;
contact the minimum number of nodes and retry if a node fails (retry means losing a significant amount of time);

The paper defines a dynamic value representing a quorum system's resilience, called f-resilience.

$f$ represents the number of failures that are tolerated by a strategy.

Quoracle computes the read and write quorums that are f-resilient using dynamic programming (DP). The logic is defined in the _f_resilient_quorums method:

"""
Original code available at: https://github.com/mwhittaker/quoracle/blob/d83a811a905a51fc13ebc00ddac036d559463380/quoracle/quorum_system.py#L276
"""

def _f_resilient_quorums(self,
                         f: int,
                         xs: List[T],
                         e: Expr) -> Iterator[Set[T]]:
    """
    Consider a set X of elements in xs. We say X is f-resilient if, despite
    removing an arbitrary set of f elements from X, X is a quorum in e.
    _f_resilient_quorums returns the set of all f-resilient quorums.
    """
    assert f >= 1

    def helper(s: Set[T], i: int) -> Iterator[Set[T]]:
        if all(e.is_quorum(s - set(failure))
               for failure in itertools.combinations(s, min(f, len(s)))):
            yield set(s)
            return

        for j in range(i, len(xs)):
            s.add(xs[j])
            yield from helper(s, j + 1)
            s.remove(xs[j])

    return helper(set(), 0)

Given an $f$ coefficient, a set of nodes, and an expression (Expr) indicating the read and write quorums, the method returns the sets of f-resilient quorums by iterating over all the possible combinations of nodes that can form a valid f-resilient quorum. Note that it is impossible to build resilient quorums in some scenarios: for example, in case f > len(xs). Therefore, the method will return an empty set.

The concept of resilience, as described in the paper, comes at the cost of capacity.

Optimal strategy search

Quoracle provides a way to search for an optimal quorum depending on a given set of parameters:

def search(nodes: List[Node[T]],
           read_fraction: Optional[Distribution] = None,
           write_fraction: Optional[Distribution] = None,
           optimize: str = LOAD,
           resilience: int = 0,
           load_limit: Optional[float] = None,
           network_limit: Optional[float] = None,
           latency_limit: Optional[datetime.timedelta] = None,
           f: int = 0,
           timeout: datetime.timedelta = datetime.timedelta(seconds=0)) \
           -> Tuple[QuorumSystem[T], Strategy[T]]:
           
           #...

The nodes parameter represents the nodes you have in the quorum system. As already discussed, the read_fraction could be a single value or a Distribution of percentages. Finally, the optimize parameter defines the objective of the optimisation.

The load_limit, network_limit, latency_limit are the additional constraints already explained in the Optimal strategy problem implementation section.

The underlying implementation of the search function uses the types we already described, such as the Strategy and theQuorumSystem classes:

def search(nodes: List[Node[T]],
           read_fraction: Optional[Distribution] = None,
           write_fraction: Optional[Distribution] = None,
           optimize: str = LOAD,
           resilience: int = 0,
           load_limit: Optional[float] = None,
           network_limit: Optional[float] = None,
           latency_limit: Optional[datetime.timedelta] = None,
           f: int = 0,
           timeout: datetime.timedelta = datetime.timedelta(seconds=0)) \
           -> Tuple[QuorumSystem[T], Strategy[T]]:
    start_time = datetime.datetime.now()

    def metric(sigma: Strategy[T]) -> float:
        if optimize == LOAD:
            return sigma.load(read_fraction, write_fraction)
        elif optimize == NETWORK:
            return sigma.network_load(read_fraction, write_fraction)
        else:
            return sigma.latency(read_fraction, write_fraction).total_seconds()

    opt_qs: Optional[QuorumSystem[T]] = None
    opt_sigma: Optional[Strategy[T]] = None
    opt_metric: Optional[float] = None

    def do_search(exprs: Iterator[Expr[T]]) -> None:
        nonlocal opt_qs
        nonlocal opt_sigma
        nonlocal opt_metric

        for reads in exprs:
            qs = QuorumSystem(reads=reads)
            if qs.resilience() < resilience:
                continue

            try:
                sigma = qs.strategy(optimize = optimize,
                                    load_limit = load_limit,
                                    network_limit = network_limit,
                                    latency_limit = latency_limit,
                                    read_fraction = read_fraction,
                                    write_fraction = write_fraction,
                                    f = f)
                sigma_metric = metric(sigma)
                if opt_metric is None or sigma_metric < opt_metric:
                    opt_qs = qs
                    opt_sigma = sigma
                    opt_metric = sigma_metric
            except NoStrategyFoundError:
                pass

            if (timeout != datetime.timedelta(seconds=0) and
                datetime.datetime.now() - start_time >= timeout):
                return

    do_search(_dup_free_exprs(nodes, max_height=2))
    do_search(_dup_free_exprs(nodes))

    if opt_qs is None:
        raise ValueError('no quorum system found')
    else:
        assert opt_sigma is not None
        return (opt_qs, opt_sigma

Given a set of nodes, the implementation calculates the list of all the de-duplicated expressions that can be composed with that set of nodes using the _dup_free_exprs function.

The implementation proceeds by calling the do_search function; it initialises a new instance of a QuorumSystem for each previously computed expression.

Once the QuorumSystem is initialised, it proceeds by calling the strategy method to get the optimal strategy for the current expression. If the target parameter for that strategy is more optimal than the previous strategy, it is promoted to an optimal strategy. Otherwise, the algorithm continues.

In terms of iterations, the do_search applies linear programming optimisation for each expression derived from the _dup_free_exprs. However, the computation is time-consuming: the search function accepts a timeout parameter limiting the search's computation time. The implementation checks at every iteration whether the timeout has been hit, in which case it stops the computation.

Final thoughts

This post went through the concepts of quorum systems illustrated in the Read-Write Quorum Systems Made Practical - Michael Whittaker, Aleksey Charapko, Joseph M. Hellerstein, Heidi Howard, Ion Stoica paper, and it gave a detailed look at the Quoracle implementation. Quoracle is an excellent foundation to explore the trade-offs of the read-write quorum systems. A closer look at the Quoracle implementation helps to understand the critical aspects of implementing quorums in distributed systems.

I have implemented a Golang port of Quoracle for learning purposes, available at samueleresca/quoracle-go. A closer look at the Golang implementation is available in the second part of this series.

Below are the references used to write this post.