from typing import Dict from paxos_utils import Address, BallotValuePair, BallotNumber class Message: def __init__(self, addr: Address) -> None: self.addr: Address = addr # address of sender def __str__(self) -> str: return f"addr: {str(self.addr)}" class LeaderHeartbeat(Message): def __init__(self, addr: Address, log: Dict[int, BallotValuePair], ballot_num: BallotNumber) -> None: super().__init__(addr) self.log: Dict[int, BallotValuePair] = log self.ballot_num: BallotNumber = ballot_num def __str__(self) -> str: return f"LeaderHeartbeat({super().__str__()}, {str(self.ballot_num)}, log: {self.log})" class P1A(Message): def __init__(self, addr: Address, ballot_num: BallotNumber) -> None: super().__init__(addr) # ballot_num is a BallotNumber self.ballot_num: BallotNumber = ballot_num def __str__(self) -> str: return f"P1A({super().__str__()}, {str(self.ballot_num)})" class P1B(Message): def __init__(self, addr: Address, accepted_ballot: BallotNumber, accepted: Dict[int, BallotValuePair]) -> None: super().__init__(addr) # accepted_ballot is a BallotNumber # accepted is a dict from int to BallotValuePair self.accepted_ballot: BallotNumber = accepted_ballot self.accepted: Dict[int, BallotValuePair] = accepted def __str__(self) -> str: return f"P1B({super().__str__()}, {str(self.accepted_ballot)}, {str(self.accepted)})" class P2A(Message): def __init__(self, addr: Address, ballot_num: BallotNumber, slot_num: int, value) -> None: super().__init__(addr) # ballot is BallotNumber # slot_num is an int # value is a PaxosRequest self.ballot_num: BallotNumber = ballot_num self.slot_num: int = slot_num self.value: PaxosRequest = value def __str__(self) -> str: return f"P2A({super().__str__()}, {str(self.ballot_num)}, slot_num: {str(self.slot_num)}, value: {str(self.value)})" class P2B(Message): def __init__(self, addr: Address, ballot: BallotNumber, slot_num: int) -> None: super().__init__(addr) # ballot is BallotNumber # slot_num is an int self.ballot_num: BallotNumber = ballot self.slot_num: int = slot_num def __str__(self) -> str: return f"P2B({super().__str__()}, {str(self.ballot_num)}, slot_num: {str(self.slot_num)})" class PaxosRequest(Message): def __init__(self, addr: Address, lock_command) -> None: super().__init__(addr) # lock_command is of type LockCommand self.lock_command: LockCommand = lock_command def __str__(self) -> str: return f"PaxosRequest({super().__str__()}, {str(self.lock_command)})" class PaxosResult(Message): def __init__(self, addr: Address, value: bool, lock_command) -> None: super().__init__(addr) self.value: bool = value self.cmd: LockCommand = lock_command def __str__(self) -> str: return f"PaxosResult({super().__str__()}, value: {str(self.value)}, cmd: {str(self.cmd)})" class LockCommand: def __init__(self, op, value, req_num) -> None: # op is either "lock" or "unlock" # value is value to lock (int) self.op = op self.value = value self.req_num = req_num def __str__(self) -> str: return f"LockCommand(op: {str(self.op)}, value: {str(self.value)})"