from functools import total_ordering from enum import Enum from typing import Tuple Address = Tuple[str, int] # Type declaration for what an address is @total_ordering class BallotNumber: def __init__(self, seq_num, addr) -> None: self.seq_num = seq_num # Sequence number of the proposal self.addr = addr # Address of the server proposing def increaseBallot(self): self.seq_num += 1 def _is_valid_operand(self, other): return (hasattr(other, "seq_num") and hasattr(other, "addr")) # Only need to make lt and eq, total_ordering will do the rest def __lt__(self, other) -> bool: if not self._is_valid_operand(other): return NotImplemented if self.seq_num < other.seq_num: return True elif self.seq_num > other.seq_num: return False else: # Same seq_num, use address for break return self.addr[1] < other.addr[1] def __eq__(self, other) -> bool: if not self._is_valid_operand(other): return NotImplemented return ((self.seq_num, self.addr) == (other.seq_num, other.addr)) def __str__(self) -> str: return "BallotNumber(seq_num: " + str(self.seq_num) + ", addr: " + str(self.addr) + ")" @total_ordering class BallotValuePair: def __init__(self, ballot_num, value) -> None: self.ballot_num = ballot_num # ballot number self.value = value # paxos request def _is_valid_operand(self, other): return (hasattr(other, "ballot_num") and hasattr(other, "value")) # Only need to make lt and eq, total_ordering will do the rest def __lt__(self, other) -> bool: if not self._is_valid_operand(other): return NotImplemented return self.ballot_num < other.ballot_num def __eq__(self, other) -> bool: if not self._is_valid_operand(other): return NotImplemented return self.ballot_num == other.ballot_num def __str__(self) -> str: return "BallotValuePair(ballot_num: " + str(self.ballot_num) + ", value: " + str(self.value) + ")" class PaxosLogSlotStatus(Enum): EMPTY = 0 # no command is known by the server for this slot CHOSEN = 1 # the server knows a command to be permanently chosen for this slot