拜占庭将军问题

"""
Byzantine Generals Problem, Oral Message(OM) Algorithm
Author: expye(Zihao Ye)
"""
from typing import List, Tuple
import argparse
import numpy as np
from collections import Counter


def majority(lst: List):
    """Note that we should output the same value for [0, 0, 1, 1] and [1, 1, 0, 0]"""
    c = Counter(sorted(lst))
    return c.most_common()[0][0]


class General:
    def __init__(self, truthful: bool) -> None:
        self._truthful = truthful

    def __call__(self, input):
        """If truthful, always output input, otherwise output random value."""
        if self._truthful:
            return input
        else:
            return np.random.randint(2)


class OralMessage:
    def __init__(self, generals: List[General], n: int, k: int) -> None:
        """
        Parameters
        ----------
        generals : List[General]
            The general list.
        n : int
            Number of generals.
        k : int
            Number of traitors.
        """
        self.generals = generals
        self.n = n
        self.k = k
        # record traces: (path, command, remaining number of rounds)
        self.trace = []
        self.received = {}  # received commands of each path started from root
        self.vote = {}      # final vote of each path

    def propagate(self, ancestors: Tuple[int], command: bool, m: int):
        """
        Regard current node as commander, propagate message for $m$ rounds, and collect traces.

        Parameters
        ----------
        ancestors : Tuple[int]
            The message passing path from root to current node.
        command : bool
            The command current node received.
        m : int
            The remaining number of rounds.
        """
        self.trace.append((ancestors, command, m))  # collect traces.
        if m > 0:  # not in the last round
            commander = ancestors[-1]
            for i in range(self.n):
                if not i in ancestors:  # send messages to other generals
                    self.propagate(ancestors + (i,),
                                   self.generals[commander](command), m - 1)

    def gather(self, ordered_trace: List[Tuple[Tuple[int], bool, int]]):
        """
        Back propagation and take majority of received commands of each path as its vote.

        Parameters
        ----------
        ordered_traces : List[Tuple[Tuple[int], bool, int]]
            Collected traces sorted by topological order.
        """
        for ancestors, command, m in ordered_trace[:-1]:
            commander = ancestors[-1]
            if m == 0:  # in the last round, take the command received in forward phase.
                vote = self.generals[commander](command)
            # take majority of received (in forward/backward phase) commands
            else:
                vote = self.generals[commander](
                    majority(self.received[ancestors] + [command]))
            self.vote[ancestors] = vote
            if len(ancestors) > 1:  # update received commands for m-1 level paths
                j = ancestors[: -2] + (commander,)
                if not j in self.received:
                    self.received[j] = []
                self.received[j].append(vote)


if __name__ == '__main__':
    parser = argparse.ArgumentParser("Byzantine generals simulator.")
    parser.add_argument('--n', '-n', type=int, help='number of generals')
    parser.add_argument('--m', '-m', type=int, help='number of runs')
    parser.add_argument('--k', '-k', type=int, help='number of traitors')
    args = parser.parse_args()
    traitors = np.random.choice(args.n, args.k)
    is_truthful = np.ones(args.n, dtype=np.int32)
    is_truthful[traitors] = 0
    generals = []
    for i in range(args.n):
        generals.append(General(is_truthful[i]))

    om = OralMessage(generals, args.n, args.k)
    om.propagate((0,), 1, args.m + 1)
    om.gather(sorted(om.trace, key=lambda _: _[2]))

    print('t' if is_truthful[0] else 'f')
    for i in range(1, args.n):
        print('t' if is_truthful[i] else 'f', end='')
    print()

    for i in range(1, args.n):
        print(om.vote[(0, i)], end='')
    print()


"""
example: python byzantine.py -n 10 -k 3 -m 3
output:
t
tttftttff
111011101

The commander is a loyal general, and all loyal generals made the same decision: 1.
"""