Skip to content

Instantly share code, notes, and snippets.

Avatar

Samuele Resca samueleresca

View GitHub Profile
View quoracle\search.py
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)) \
View quoracle\quorum_system.py
"""
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
View quoracle\strategy.py
"""
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:
View quoracle\strategy.py
"""
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
View quoracle\strategy.py
"""
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
View quoracle\quorum_system.py
"""
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(
View quoracle\quorum_system.py
"""
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)
View quoracle\quorum_system.py
'''
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())
View quoracle\distribution.py
'''
Original code available at: https://github.com/mwhittaker/quoracle
'''
from typing import Dict, Optional, Union
Fraction = float
Weight = float
Probability = float
Distribution = Union[
View quoracle\quorum_system.py
"""
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(