samueleresca

"""
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:

samueleresca

"""
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

samueleresca

"""
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

samueleresca

"""
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(

samueleresca

"""
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)

samueleresca

'''
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())

samueleresca

'''
Original code available at: https://github.com/mwhittaker/quoracle
'''

from typing import Dict, Optional, Union

Fraction = float
Weight = float
Probability = float
Distribution = Union[

samueleresca

"""
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(

samueleresca

"""
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])

samueleresca

"""
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: