/* Fault-tolerant consensus, W.H. Hesselink, 2nd October 2003.
Version 12 October 2008. 
See the third problem of 
P. Berman, J.A. Garay: 
Asymptotically optimal distributed consensus (extended abstract).
In G. Ausiello, M. Denazi-Ciancaglini, S. Ronchi Della Rocca (eds.):
Automata, Languages and Programming. LNCS 372, pp. 80--94 (1989).

This is an example of so-called Byzantine agreement. The tale is that 
some generals of the realm of Byzantium in the Middle Ages were unreliable.
Therefore the Emperor needed protocols to enforce agreement among them 
in spite of the unreliable ones. 

We assume that the unreliable processes act according to the protocol, 
but send arbitrary values and set their own val[self] arbitrarily.

In the second phases of the rounds, not all reliable processes play
the central role, but it is guaranteed that the central role is played
by a reliable process at least once.
*/

// Set physical memory to 512 (Set Advanced Options). Compression also helps.

#define NrOfRounds     2
#define UnReliable     1 /* Number of unreliable generals */
#define NrOfGens       5 /* Number of generals */
#define ReliableSet   30 /* Bit pattern of the reliables; here {1, 2, 3, 4} */
  // The most demanding case: first an unreliable round, then a reliable one.
  //  29 (a reliable round followed by an unreliable one) needs less memory.

//  The case of 4 generals with 1 unreliable one need not reach consensus:
//#define NrOfGens     4 /* Number of generals */
//#define ReliableSet 13 /* Bit pattern of the reliables; here {0, 2, 3} */
 
#define IsReliable(self)  (1 & (ReliableSet >> (self)))
#define ConsensusPos      ((val & ReliableSet) == ReliableSet)
#define ConsensusNeg      ((val & ReliableSet) == 0)
#define InitConsensusPos  ((initVal & ReliableSet) == ReliableSet)
#define InitConsensusNeg  ((initVal & ReliableSet) == 0)

typedef ChanArr {
    chan ch[NrOfGens] = [1] of {bit} ; 
}

ChanArr chans [NrOfGens] ;
byte val, initVal ;
byte nr[NrOfGens] ;

inline broadcast(self, x) {
        atomic { 
            do 
            :: i < NrOfGens && ! IsReliable(self) ->  // unreliable
                 chans[self].ch[i] ! !x ; 
                 i ++ 
            :: i < NrOfGens -> 
                 chans[self].ch[i] ! x ;
                 i ++ 
            :: i == NrOfGens -> break 
            od ;
            i = 0
        } 
}

inline reception(self) {    /* in first phase */
        nr[self] = 0 ;
        do                            
        :: i < NrOfGens ->
            d_step {
               chans[i].ch[self] ? nieuw ; 
               nr[self] = (IsReliable(self) -> nr[self] + nieuw : 0) ; 
               nieuw = 0 ;
               i++  
           }
        :: i == NrOfGens -> break 
        od ;
        i = 0 ;
}

inline updateOwnValue(self) {
         d_step {
              chans[round].ch[self] ? nieuw ;  /* receiving in second phase */
              if 
              :: IsReliable(self) ->           /* update val[self] */
                   nieuw = (nr[self] <= UnReliable -> 0 :
                            (nr[self] >= NrOfGens - UnReliable -> 1 : 
                             nieuw  )) ;
                   val = (nieuw -> val | (1 << self) : val & ~(1 << self)) 
              :: else skip 
              fi ;
          }
}

init {
    byte round = 0, i = 0 , self = 0 ;
    bit nieuw ;
    val = 0 ;
    do 
    :: val < (1 << NrOfGens) - 1 -> val ++
    :: break 
    od ;  // val is chosen arbitrarily < (1<<NrOfGens).
    initVal = val ;
    do 
    :: round < NrOfRounds -> 
         do
         :: self < NrOfGens -> 
              broadcast(self, 1 & (val >> self)) ; /* own value */
              self ++ ;
         :: self == NrOfGens -> self = 0 ; break
         od ;
         do 
         :: self < NrOfGens -> 
              reception(self) ;          /* determines nr[self] */
              self ++ 
         :: else -> self = 0 ; break 
         od ;
         /* second phase */
         broadcast(round, NrOfGens <= 2*nr[round]) 
         do 
         :: self < NrOfGens -> 
              updateOwnValue(self) ; self ++ 
         :: else -> self = 0 ; break
         od ;
         round ++ 
    :: else -> break
    od ;
    atomic {
        assert ConsensusPos || ConsensusNeg ;
        assert (! InitConsensusPos || ConsensusPos) 
                && (! InitConsensusNeg || ConsensusNeg) ;
     }
}

/* Recall that there are N generals, K of which are unreliable, and that
4K < N.

Once the reliable processes have reached consensus, this consensus is
preserved because all reliables send a 0 or a 1; then all reliables
obtain in reception nr <= K or nr >= N-K, and hence keep their values
because of the first two alternatives of the update of val[self].
For this purpose, 2K < N is sufficient.

Assuming UnReliable < NrOfRounds, there is at least one value for
round that corresponds to a reliable process. In this round, consensus
is reached. This is shown as follows.

Assume that the round is the number of a reliable process i.  Consider
the precondition of this round: assume there are x reliable processes
with val[self] = 1. Then N-K-x reliable processes have 0. In the first
phase all processes receive nr in [x..x+K]. There are three cases.

1. Some reliable general gets nr <= K. Then x <= K. Therefore 
x+K <= 2K < N/2. Coordinator i therefore sends 0, and all reliable 
generals set val[self] := 0. Hence consensus.

2. Some reliable general get nr >= N-K. Then x+K >= N-K. Therefore 
x >= N-2K > N/2. Coordinator i therefore sends 1, and all reliable 
generals set val[self] := 1. Hence consensus.

3. All reliable generals set val[self] := nieuw as sent by
coordinator i. This also results in consensus.  */