graduate_code/solutions/distributed.py

"""G08: bounded combinatorial checks, not a production storage protocol."""
from dataclasses import dataclass
from itertools import combinations


def disjoint_quorums(n, read_size, write_size):
    if any(type(x) is not int for x in (n,read_size,write_size)):
        raise ValueError("strict integer sizes required")
    if not 1 <= n <= 12 or not 1 <= read_size <= n or not 1 <= write_size <= n:
        raise ValueError("bounded valid quorum sizes required")
    for readers in combinations(range(n),read_size):
        complement = [i for i in range(n) if i not in readers]
        # Enumerate write quorums disjoint from this read set only.
        for writers in combinations(complement,write_size):
            return set(readers),set(writers)
    return None


@dataclass
class LogicalClock:
    value: int = 0

    def __post_init__(self):
        if type(self.value) is not int or self.value < 0:
            raise ValueError("nonnegative integer clock required")

    def tick(self):
        self.value += 1
        return self.value

    def receive(self, timestamp):
        if type(timestamp) is not int or timestamp < 0:
            raise ValueError("nonnegative integer message timestamp required")
        self.value = max(self.value,timestamp)+1
        return self.value


def unsafe_register_trace():
    """Deliberately violates linearizable read behavior; never use as storage."""
    replicas = {name:(0,"old") for name in "ABC"}
    replicas["A"] = (1,"new")  # incomplete write, no write quorum
    first = max(replicas[name] for name in "AB")
    second = max(replicas[name] for name in "BC")
    return first,second


if __name__ == "__main__":
    print(disjoint_quorums(5,2,3))
    print(unsafe_register_trace())