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(
                    '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:
        ...