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