Shahzad Bhatti Welcome to my ramblings and rants!

public enum Value {
    ZERO,
    ONE,
    FAULTY,
    UNKNOWN
}

public interface ValueStrategy {

    public Node getSourceValue();

    public Value getValue(
            Value value,
            int source,
            int destination,
            String path);


    public Value getDefaultValue();


    public boolean isFaulty(int process);
}

public interface Broadcaster {
    public void broadcast(int round, int id, Node source, String path);
}

public interface PathRepository {

    public void generateChildren();

    public List<String> getRankList(int rank, int source);

    public List<String> getPathChildren(String path);
}

public class Configuration {
    private int source;
    private int roundsOfMessages;
    private int numberOfProcesses;

    //
    Configuration(int source, int roundsOfMessages, int numberOfProcesses) {
        this.source = source;
        this.roundsOfMessages = roundsOfMessages;
        this.numberOfProcesses = numberOfProcesses;
    }

    //
    public int getSource() {
        return source;
    }


    public int getRoundsOfMessages() {
        return roundsOfMessages;
    }


    public int getNumberOfProcesses() {
        return numberOfProcesses;
    }
}

public class DefaultEngine implements Engine, Broadcaster {
    static final boolean debug = System.getProperty("debug", "false").equals("true");
    private Configuration configuration;
    private PathRepository repository;
    private List<Process> processes;
    private ValueStrategy strategy;

    //
    public DefaultEngine(int source, int roundsOfMessages, int numberOfProcesses) {
        this.configuration = new Configuration(source, roundsOfMessages, numberOfProcesses);
        this.strategy = new DefaultValueStrategy(configuration);
        this.repository = new DefaultPathRepository(configuration);
        this.processes = new ArrayList<Process>();
    }

    public void broadcast(int round, int id, Node source, String path) {
        for (int j = 0; j < configuration.getNumberOfProcesses(); j++) {
            if (j != configuration.getSource()) {
                Value value = strategy.getValue(
                        source.input,
                        id,
                        j,
                        path);
                if (debug) {
                    String sourcePath = path.substring(0, path.length() - 1);
                    System.out.println("Sending for round " + round + " from process " + id + " to " + j + ": {" + value +
                            ", " + path + ", " + Value.UNKNOWN + "}, getting value from source " + sourcePath);
                }
                getProcesses().get(j).receiveMessage(path, new Node(value, Value.UNKNOWN));
            }
        }
    }


    public List<Process> getProcesses() {
        return processes;
    }

    public void run() {
        //
        // Starting at round 0 and working up to round M, call the
        // SendMessages() method of each process. It will send the appropriate
        // message to all other sibling processes.
        //
        for (int i = 0; i <= configuration.getRoundsOfMessages(); i++) {
            for (int j = 0; j < configuration.getNumberOfProcesses(); j++) {
                processes.get(j).sendMessages(i);
            }
        }

        //
        // All that is left is to print out the results. For non-faulty processes,
        // we call the Decide() method to see what what the process decision was
        //
        for (int j = 0; j < configuration.getNumberOfProcesses(); j++) {
            if (processes.get(j).isSource()) System.out.print("Source ");
            System.out.print("Process " + j);
            if (!processes.get(j).isFaulty()) {
                System.out.print(" decides on value " +
                        processes.get(j).decide());
            } else {
                System.out.print(" is faulty");
            }
            System.out.println();
        }
    }


    public void start() {
        //
        for (int i = 0; i < configuration.getNumberOfProcesses(); i++) {
            Process process = i == configuration.getSource() ? new GeneralProcess(i, configuration, repository, this, strategy) : new Process(i, configuration, repository, this, strategy);
            processes.add(process);
        }
        repository.generateChildren();
    }
}

public class DefaultPathRepository implements PathRepository {
    private Map<String, List<String>> children;
    private Map<Integer, Map<Integer, List<String>>> pathByRank;
    private Configuration configuration;
    static final boolean debug = System.getProperty("debug", "false").equals("true");

    //
    public DefaultPathRepository(Configuration configuration) {
        this.configuration = configuration;
        this.children = new HashMap<String, List<String>>();
        this.pathByRank = new HashMap<Integer, Map<Integer, List<String>>>();
    }

    public Map<String, List<String>> getChildren() {
        return children;
    }

    public List<String> getPathChildren(String path) {
        List<String> pathList = children.get(path);
        if (pathList == null) {
            pathList = new ArrayList<String>();
            children.put(path, pathList);
        }
        return pathList;
    }


    public List<String> getRankList(int rank, int source) {
        Map<Integer, List<String>> pathMap = pathByRank.get(rank);
        if (pathMap == null) {
            pathMap = new HashMap<Integer, List<String>>();
            pathByRank.put(rank, pathMap);
        }
        List<String> pathList = pathMap.get(source);
        if (pathList == null) {
            pathList = new ArrayList<String>();
            pathMap.put(source, pathList);
        }
        return pathList;
    }

    public void generateChildren() {
        generateChildren(configuration.getSource(), new boolean[configuration.getNumberOfProcesses()], "", 0);
    }


    private void generateChildren(int source, boolean[] ids, String path, int rank) {
        ids[source] = true;
        path += toCharString(source);
        getRankList(rank, source).add(path);

        //
        if (rank < configuration.getRoundsOfMessages()) {
            for (int i = 0; i < ids.length; i++) {
                if (!ids[i]) {
                    boolean[] newIds = new boolean[ids.length];
                    System.arraycopy(ids, 0, newIds, 0, ids.length);
                    generateChildren(i, newIds, path, rank + 1);
                    getPathChildren(path).add(path + toCharString(i));
                }
            }
        }

        //
        if (debug) {
            System.out.print("generateChildren(" + source + "," + rank + "," + path + "), children = ");
            List<String> list = getPathChildren(path);
            for (String s : list) {
                System.out.print(s + " ");
            }
            System.out.println();
        }
    }


    //
    private char toChar(int n) {
        return (char) (n + '0');
    }

    private String toCharString(int n) {
        return String.valueOf(toChar(n));
    }
}

public class DefaultValueStrategy implements ValueStrategy {
    private Configuration configuration;
    private Node sourceValue;

    //
    DefaultValueStrategy(Configuration configuration) {
        this.configuration = configuration;
        this.sourceValue = new Node(Value.ZERO, Value.UNKNOWN);
    }

    //
    // This method returns the true value of the source's value. The source may send
    // faulty values to other processes, but the Node returned by this value will be
    // in its root node.
    //
    // In this case, the General's node has in input value of 0, which makes that
    // the desired value. Of course, since the General is faulty, this doesn't really
    // matter.
    //
    public Node getSourceValue() {
        return sourceValue;
    }

    //
    // During message, GetValue() is called to get the value returned by a given process
    // during a messaging round. 'value' is the input value that it should be sending to
    // the destination process (if it isn't faulty), source is the source process ID,
    // destination is the destination process ID, and Path is the path being used for this
    // particular message.
    //
    // In this particular implementation, we have two faulty processes - the source
    // process, which returns a sort-of random value, and process ID 2, which returns
    // a ONE always, in contradiction of the General's desired value of 0.
    //
    public Value getValue(Value value, int source, int destination, String path) {
        if (configuration.getSource() == source) return (destination & 1) != 0 ? Value.ZERO : Value.ONE;
        else if (source == 2) return Value.ONE;
        return value;
    }


    //
    // When breaking a tie, GetDefault() is used to get the default value.
    //
    // This is an arbitrary decision, but it has to be consistent across all processes.
    // More importantly, the processes have to arrive at a correct decision regardless
    // of whether the default value is always 0 or always 1. In this case we've set it to
    // a value of 1.
    //
    public Value getDefaultValue() {
        return Value.ONE;
    }

    //
    // This method is used to identify fault processes by ID
    //
    public boolean isFaulty(int process) {
        return process == configuration.getSource() || process == 2;
    }
}

public interface Engine extends Runnable {
    public void start();
}

public class GeneralProcess extends Process {
    public GeneralProcess(int id, Configuration configuration, PathRepository repository, Broadcaster broadcaster, ValueStrategy strategy) {
        super(id, configuration, repository, broadcaster, strategy);
        nodes.put("", strategy.getSourceValue());
    }


    @Override
    public Value decide() {
        //
        // The source process doesn't have to do all the work - since it's the decider,
        // it simply looks at its input value to pick the appropriate decision value.
        //
        return nodes.get("").input;
    }
}

public class Node {
    Value input;
    Value output;

    Node() {
        this(Value.FAULTY, Value.FAULTY);
    }

    Node(Value input, Value output) {
        this.input = input;
        this.output = output;
    }
}

public class Process {
    protected int id;
    protected Configuration configuration;
    protected ValueStrategy strategy;
    protected Map<String, Node> nodes;
    protected PathRepository repository;
    protected Broadcaster broadcaster;
    static final boolean debug = System.getProperty("debug", "false").equals("true");

    public Process(int id, Configuration configuration, PathRepository repository, Broadcaster broadcaster, ValueStrategy strategy) {
        this.id = id;
        this.strategy = strategy;
        this.repository = repository;
        this.broadcaster = broadcaster;
        this.configuration = configuration;
        this.nodes = new HashMap<String, Node>();
    }

    static int totalMessages;

    //
    //
    // Receiving a message is pretty simple here, it means that some other process
    // calls this method on the current process with path and a node. All we do
    // is store the value, we'll use it in the next round of messaging.
    //
    public void receiveMessage(String path, Node node) {
        nodes.put(path, node);
        totalMessages++;
    }


    //
    // After constructing all messages, you need to call SendMessages on each process,
    // once per round. This routine will send the appropriate messages for each round
    // to all th eother processes listed in the vector passed in as an argument.
    //
    // Deciding on what messages to send is pretty simple. If we look at the static
    // map pathsByRank, indexed by round and the processId of this process, it gives
    // the entire set of taraget paths that this process needs to send messages to.
    // So there is an iteration loop through that map, and this process sends a message
    // to the correct target process for each path in the map.
    //
    public void sendMessages(int round) {
        List<String> pathList = repository.getRankList(round, id);
        for (String path : pathList) {
            String sourcePath = path.substring(0, path.length() - 1);
            Node source = nodes.get(sourcePath);
            if (source == null) throw new IllegalStateException("Failed to find source node for " + sourcePath);
            broadcaster.broadcast(round, id, source, path);
        }
    }

    //
    // After all messages have been sent, it's time to Decide.
    //
    // This part of the algorithm follows the description in the article closely.
    // It has to work its way from the leaf values up to the root of the tree.
    // The first round consists of going to the leaf nodes, and copying the input
    // value to the output value.
    //
    // All subsequent rounds consist of getting the majority of the output values from
    // each nodes children, and copying that to the nodes output value.
    //
    // When we finally reach the root node, there is only one node with an output value,
    // and that represents this processes decision.
    //
    public Value decide() {
        //
        // Step 1 - set the leaf values
        //
        for (int i = 0; i < configuration.getNumberOfProcesses(); i++) {
            List<String> pathList = repository.getRankList(configuration.getRoundsOfMessages(), i);
            for (String path : pathList) {
                Node node = nodes.get(path);
                node.output = node.input;
            }
        }
        //
        // Step 2 - work up the tree
        //
        for (int round = configuration.getRoundsOfMessages() - 1; round >= 0; round--) {
            for (int i = 0; i < configuration.getNumberOfProcesses(); i++) {
                List<String> pathList = repository.getRankList(round, i);
                for (String path : pathList) {
                    Node node = nodes.get(path);
                    node.output = getMajority(path);
                }
            }
        }
        List<String> pathList = repository.getRankList(0, configuration.getSource());
        String topPath = pathList.get(0);
        return nodes.get(topPath).output;
    }

    //
    // This routine calculates the majority value for the children of a given
    // path. The logic is pretty simple, we increment the count for all possible
    // values over the children. If there is a clearcut majority, we return that,
    // otherwise we return the default value defined by the strategy class.
    //
    public Value getMajority(String path) {
        Map<Value, Integer> counts = new HashMap<Value, Integer>();
        counts.put(Value.ONE, new Integer(0));
        counts.put(Value.ZERO, new Integer(0));
        counts.put(Value.UNKNOWN, new Integer(0));
        Collection<String> list = repository.getPathChildren(path);
        int n = 0;
        if (list == null) {
            if (debug) System.out.println("No child found for '" + path + "'");
        } else {
            n = list.size();
            for (String child : list) {
                Node node = nodes.get(child);
                if (node != null) {
                    counts.put(node.output, counts.get(node.output) + 1);
                } else if (debug) {
                    //System.out.println("Could not find node for count with path " + child);
                }
            }
        }
        //
        //
        if (counts.get(Value.ONE) > (n / 2)) return Value.ONE;
        else if (counts.get(Value.ZERO) > (n / 2)) return Value.ZERO;
        else if (counts.get(Value.ONE).intValue() == counts.get(Value.ZERO).intValue() &&
                counts.get(Value.ONE) == (n / 2)) return strategy.getDefaultValue();
        return Value.UNKNOWN;
    }


    //
    // A utility routine that tells whether a given process is faulty
    //
    public boolean isFaulty() {
        return strategy.isFaulty(id);
    }

    //
    // Another somewhat handy utility routine
    //
    public boolean isSource() {
        return configuration.getSource() == id;
    }
}

class Configuration
	attr_reader :source, :num_rounds, :num_procs
	def initialize(source, num_rounds, num_procs)
		@source = source
		@num_rounds = num_rounds
		@num_procs = num_procs
	end
	def to_s
		"#{@source}|#{@num_rounds}|#{@num_procs}"
	end
end

class Node
	attr_accessor :input
	attr_accessor :output
	def initialize(input = Value::FAULTY, output = Value::FAULTY)
		@input = input
		@output = output
	end
	def to_s
		"#{@input}/#{@output}"
	end
end

class Value
	ZERO = '0'
	ONE = '1'
	FAULTY = 'X'
	UNKNOWN = '?'
end

class PathRepository
	attr_accessor :children
	attr_accessor :path_by_rank
	attr_accessor :config
		def initialize(config)
		@config = config
		@children = Hash.new {|h,k| h[k] = []}
		@path_by_rank = Hash.new {|h,k| h[k] = Hash.new {|hh,kk| hh[kk] = []}}
	end
		def path_children(path)
		@children[path]
	end
		def rank_list(rank, source)
		@path_by_rank[rank][source]
	end
		def generate_children(source = @config.source, ids = new_ids, path = "", rank = 0)
		ids[source] = true
		path = "#{path}#{source}"
		@path_by_rank[rank][source].push path
				if rank < @config.num_rounds
			for i in 0...@config.num_procs
			unless ids[i]
				generate_children(i, new_ids(ids), path, rank + 1)
				@children[path].push("#{path}#{i}")
			end
		end
	end
		if (DEBUG)
		print("generate_children(#{source}, #{rank}, #{path}, children = ")
		@children[path].each do |child|
			print(child + " ")
		end
		puts("")
	end
end
private
def new_ids(old_ids = nil)
	#old_ids.nil? ? Array.new(@config.num_procs) {|i| false} : Array.new(old_ids)
	old_ids.nil? ? Hash.new {|h,k| h[k] = false} : old_ids.dup
end
end

class ValueStrategy
	attr_accessor :config
	attr_accessor :source_value
		def initialize(config)
		@config = config
		@source_value = Node.new(Value::ZERO, Value::UNKNOWN)
	end
		def create_value(value, source, destination, path)
		if @config.source == source
			(destination & 1) != 0 ? Value::ZERO : Value::ONE
		elsif source == 2
			Value::ONE
		else
			value
		end
	end
			def get_default
		return Value::ONE
	end
		def faulty?(process)
		process == @config.source || process == 2
	end
end
class LProcess
	@@total_messages = 0
	attr_reader :id
	attr_reader :config
	attr_reader :strategy
	attr_reader :repository
	attr_reader :broadcaster
	#
	def initialize(id, config, repository, broadcaster, strategy)
		@id = id
		@config = config
		@repository = repository
		@strategy = strategy
		@broadcaster = broadcaster
		@nodes = Hash.new
	end
			def self.total_messages
		@@total_messages
	end
	def self.reset_total_messages
		@@total_messages = 0
	end
		def receive_message(path, node)
		@nodes[path] = node
		@@total_messages += 1
	end
		def send_messages(round)
		@repository.rank_list(round, @id).each do |path|
			source_path = path.slice(0, path.length-1)
			source = @nodes[source_path]
			raise "Source path #{source_path} not found" if source.nil?
			broadcaster.broadcast(round, @id, source, path)
		end
	end
		def decide
		#### Step 1 - set the leaf values
		for i in 0...@config.num_procs
		@repository.rank_list(@config.num_rounds, i).each do |path|
			node = @nodes[path]
			node.output = node.input
		end
	end
		### Step 2 - work up the tree
	(@config.num_rounds - 1).step 0, -1 do |round|
		for i in 0...@config.num_procs
		@repository.rank_list(round, i).each do |path|
			@nodes[path].output = find_majority(path)
		end
	end
end
path_list = @repository.rank_list(0, @config.source)
top_path = path_list[0]
@nodes[top_path].output
end
def find_majority(path)
	counts = Hash.new(0)
	list = @repository.path_children(path)
	n = 0
	if list
		n = list.size
		list.each do |child|
			node = @nodes[child]
			counts[node.output] = counts[node.output] + 1 if node
		end
	end
	#
	if (counts[Value::ONE] > ( n / 2 ) )
		Value::ONE
	elsif (counts[Value::ZERO] > ( n / 2 ) )
		Value::ZERO
	elsif (counts[Value::ONE] == counts[Value::ZERO] &&
	counts[Value::ONE] == (n / 2))
	@strategy.get_default
else
	Value::UNKNOWN
end
end
def faulty?
	@strategy.faulty?(id)
end
def source?
	@config.source == id
end
end
class GeneralProcess < LProcess
	def initialize(id, config, repository, broadcaster, strategy)
		super(id, config, repository, broadcaster, strategy)
		@nodes[""] = @strategy.source_value
	end
			def decide
		@nodes[""].input
	end
end

class Engine
	attr_reader :config
	attr_reader :repository
	attr_reader :procs
	attr_reader :strategy
		def initialize(source, num_rounds, num_procs)
		@config = Configuration.new(source, num_rounds, num_procs)
		@strategy = ValueStrategy.new(config)
		@repository = PathRepository.new(config)
		@procs = []
	end
		def broadcast(round, id, source, path)
		for j in 0...@config.num_procs
		unless j == @config.source
			value = @strategy.create_value(source.input, id, j, path)
			if (DEBUG)
				source_path = path.slice(0, path.length-1)
				puts("Sending for round #{round} from process #{id} to #{j} : #{value}, #{path}, #{Value::UNKNOWN}, getting value from source #{source_path}")
			end
			@procs[j].receive_message(path, Node.new(value, Value::UNKNOWN))
		end
	end
end
def run
	for i in 0...@config.num_rounds
	for j in 0...@config.num_procs
	@procs[j].send_messages(i)
end
end
for j in 0...@config.num_procs
print("Source ") if @procs[j].source?
print("Process #{j}")
unless @procs[j].faulty?
	print("decides on value #{@procs[j].decide}")
else
	print(" is faulty")
end
puts("")
end
end
def start
	for i in 0...@config.num_procs
	process = i == @config.source ?
	GeneralProcess.new(i, @config, @repository, self, @strategy) :
	LProcess.new(i, @config, @repository, self, @strategy)
	@procs.push(process)
end
@repository.generate_children
end
end
class ByzGeneralTest < Test::Unit::TestCase
	def setup
	end
			def run_engine(m, n, source)
		r = Benchmark.realtime() do
			puts("Starting|#{m}|#{n}|#{source}")
			engine = Engine.new(source, m, n)
			engine.start
			engine.run
		end
		puts("Finished|#{m}|#{n}|#{source}|#{LProcess.total_messages}|#{r*1000}")
		Process.reset_total_messages
	end
			def xtest_once
		n = 5
		m = 6
		source = 3
		run_engine(m, n, source)
	end
		def test_multiple
		5.step 50, 5 do |n|
			10.step 100, 10 do |m|
				source = n / 3
				run_engine(m, n, source)
			end
		end
	end
end
Test::Unit::UI::Console::TestRunner.run(ByzGeneralTest)

-module(config).
-compile(export_all).


-record(config, {source, num_procs, num_rounds}).

new(S, N, M) ->
  #config{source = S, num_procs = N, num_rounds = M}.

get_source(Self) ->
  Self#config.source.
get_num_procs(Self) ->
  Self#config.num_procs.
get_num_rounds(Self) ->
  Self#config.num_rounds

-module(counter).
-export([start/0, get_value/0, increment/0, reset/0, die/0, test_counter/0]).
-include("byz_general_test.hrl").

start() ->
  register(message_counter, spawn(fun() -> loop(0) end)).


get_value() ->
  lib_util:rpc(message_counter, value).


increment() ->
  message_counter ! {self(), increment}.

reset() ->
  message_counter ! {self(), reset}.


die() ->
  message_counter ! {self(), die, "Exiting"}.



loop(N) ->
  receive
    {_From, increment} ->
      loop(N + 1);
    {_From, reset} ->
      loop(0);
    {From, value} ->
      From ! {message_counter, N},
      loop(N);
    {_From, die, Reason} ->
      unregister(message_counter),
      exit(Reason);
    Any ->
      io:format("Unexpected message ~p~n", [Any])
  end.



test_counter() ->
  start(),
  increment(),
  increment(),
  2 = get_value(),
  reset(),
  0 = get_value(),
  die()

-module(engine).
-export([start/3, init/0, run/0, reset/0, test/0]).

init() ->
  counter:start(),
  repository:start().

reset() ->
  counter:reset(),
  repository:reset().

start(Source, N, M) ->
  Config = config:new(Source, N, M),
  put(config, Config),
  ProcIds = lists:seq(0, config:get_num_procs(Config) - 1),
  Pids = lists:map(fun(I) -> process:start(I, Config) end, ProcIds),
  put(pids, Pids),
  lists:foreach(fun(Pid) -> process:init(Pid, Pids) end, Pids),
  repository:generate_children(Config),
  Config.

run() ->
  Config = get(config),
  Ids = lists:seq(0, config:get_num_procs(Config) - 1),
  lists:foreach(fun(Id) -> send_message(Id) end, Ids),
  lists:foreach(fun(Id) -> print_result(Id) end, Ids).

get_pid(Id) ->
  Pids = get(pids),
  lists:nth(Id + 1, Pids).

send_message(Id) ->
  Config = get(config),
  RoundIds = lists:seq(0, config:get_num_rounds(Config) - 1),
  lists:foreach(fun(Round) -> send_message(Id, Round) end, RoundIds).

send_message(Id, Round) ->
  Pid = get_pid(Id),
  process:send_messages(Pid, Round, Id).

print_result(Id) ->
  Config = get(config),
  Source = config:get_source(Config),
  case Source of
    Id ->
      io:format("Source ");
    _ ->
      true
  end,
  io:format("Process ~p ", [Id]),

  Pid = get_pid(Id),
  Faulty = process:is_faulty(Pid),
  if
    Faulty ->
      io:format(" is faulty~n");
    true ->
      Decision = process:decide(Pid),
      io:format(" decides on value ~p~n", [Decision])
  end.

test() ->
  init(),
  Config = start(3, 5, 6),
  run(),
  Config.

-module(lib_util).
-compile(export_all).

rpc(Pid, Request) ->
  Pid ! {self(), Request},
  receive
    {Pid, Response} ->
      Response
  end.

-module(node).
-include("byz_general_test.hrl").
-compile(export_all).

-record(node, {input, output}).

new() ->
  #node{input = ?FAULTY, output = ?FAULTY}.
new(I, O) ->
  #node{input = I, output = O}.


set_input(Self, I) ->
  Self#node{input = I}.
get_input(Self) ->
  Self#node.input.

set_output(Self, O) ->
  Self#node{output = O}.
get_output(Self) ->
  Self#node.output.
set_output_as_input(Self) ->
  O = get_output(Self),
  Self#node{output = O}.

-module(process).
-export([init/2, start/2, receive_message/3, send_messages/3, decide/1, find_majority/2, is_faulty/1, is_source/1, test_process/0]).
-include("byz_general_test.hrl").


%
%%% starts process loop
%
start(Id, Config) ->
  spawn(fun() -> loop(Id, Config) end).

receive_message(Pid, Path, Node) ->
  %lib_util:rpc(Pid, {receive_message, Path, Node}).
  Pid ! {self(), {receive_message, Path, Node}}.

send_messages(Pid, Round, Id) ->
  %lib_util:rpc(Pid, {send_messages, Round, Id}).
  Pid ! {self(), {send_messages, Round, Id}}.

init(Pid, Pids) ->
  Pid ! {self(), init, Pids}.

decide(Pid) ->
  lib_util:rpc(Pid, decide).

find_majority(Pid, Path) ->
  lib_util:rpc(Pid, {find_majority, Path}).

is_faulty(Pid) ->
  lib_util:rpc(Pid, is_faulty).

is_source(Pid) ->
  lib_util:rpc(Pid, is_source).

put(config, Config),
put(allPids, AllPids),
SourcePid = lists:nth(config:get_source(Config)+ 1, AllPids),
put(lieutenants, AllPids -- [SourcePid]),
Nodes = dict:new(),
put(nodes, Nodes),
Source = config:get_source(Config),
SourceValue = strategy:source_value(),
if
Source =:= Id ->
Nodes1 = dict:store("", SourceValue, Nodes),
put(nodes, Nodes1);
true ->
true
end .


%
%%% stores message receievd in nodes dictionary
%%% we are also keeping track of total number of messages received in simulation.
%
receive_message(Path, Node) ->
  Nodes = get(nodes),
  Nodes1 = dict:store(Path, Node, Nodes),
  put(nodes, Nodes1),
  counter:increment().

get_node(Path) ->
  Nodes = get(nodes),
  Response = dict:find(Path, Nodes),
  case Response of
    {ok, Node} ->
      Node;
    _ ->
      %io:format("Failed to find node for path ~p~n", [Path]),
      node:new()
  end.

%
%%%
%
psend_messages(Round, Id) ->
  L = repository:get_rank_list(Round, Id),
  lists:foreach(
    fun(Path) ->
      psend_messages(Round, Id, Path) end, L).

psend_messages(Round, Id, Path) ->
  Length = string:len(Path),
  SourcePath = if
                 Length > 1 ->
                   string:substr(Path, 1, Length - 1);
                 true ->
                   ""
               end,
  Source = get_node(SourcePath),
  broadcast(Round, Id, Source, Path).


%
%%% broadcast message to all processes
%
broadcast(Round, Id, SourceNode, Path) ->
  Config = get(config),
  ProcIds = lists:seq(0, config:get_num_procs(Config) - 1),
  IdsToProcess = ProcIds -- [config:get_source(Config)],
  lists:foreach(
    fun(Dest) ->
      broadcast(Round, Id, SourceNode, Path, Dest) end, IdsToProcess).

broadcast(_Round, Id, SourceNode, Path, Dest) ->
  Config = get(config),
  Value = strategy:create_value(Config, node:get_input(SourceNode), Id, Dest, Path),
  ProcPids = get(allPids),
  ProcPid = lists:nth(Dest + 1, ProcPids),
  receive_message(ProcPid, Path, node:new(Value, ?UNKNOWN)).


decide() ->
  %%% Step 1 - set the leaf values
  Config = get(config),
  ProcIds = lists:seq(0, config:get_num_procs(Config) - 1),
  lists:foreach(fun(X) -> reset_node(X) end, ProcIds),
  %%% Step 2 - work up the tree
  RoundIds = lists:reverse(lists:seq(0, config:get_num_rounds(Config) - 1)),
  lists:foreach(
    fun(Round) ->
      set_node_value(Round) end, RoundIds),
  Source = config:get_source(Config),
  Result = repository:get_rank_list(0, Source),
  case Result of
    [] ->
      ?UNKNOWN;
    [TopPath] ->
      node:get_output(get_node(TopPath));
    [TopPath | _] ->
      node:get_output(get_node(TopPath))
  end.


set_node_value(Round) ->
  Config = get(config),
  ProcIds = lists:seq(0, config:get_num_procs(Config) - 1),
  lists:foreach(
    fun(Id) ->
      set_node_value(Round, Id) end, ProcIds).

set_node_value(Round, Id) ->
  L = repository:get_rank_list(Round, Id),
  lists:foreach(
    fun(Path) ->
      set_node_value(Round, Id, Path) end, L).

set_node_value(_Round, _Id, Path) ->
  Nodes = get(nodes),
  Node = get_node(Path),
  Value = find_majority(Path),
  Nodes1 = dict:store(Path, node:set_output(Node, Value), Nodes),
  put(nodes, Nodes1).

reset_node(I) ->
  Config = get(config),
  Nodes = get(nodes),
  L = repository:get_rank_list(config:get_num_rounds(Config), I),
  lists:foreach(fun(Path) ->
    Node = get_node(Path),
    Nodes1 = dict:store(Path, node:set_output_as_input(Node), Nodes),
    put(nodes, Nodes1)
                end, L).


increment_count(Child) ->
  Counts = get(counts),
  Node = get_node(Child),
  Count = dict:fetch(node:get_output(Node), Counts) + 1,
  Counts1 = dict:store(node:get_output(Node), Count, Counts),
  put(counts, Counts1).

find_majority(Path) ->
  Counts = dict:new(),
  Counts1 = dict:store(?ONE, 0, Counts),
  Counts2 = dict:store(?ZERO, 0, Counts1),
  Counts3 = dict:store(?UNKNOWN, 0, Counts2),
  put(counts, Counts3),
  L = repository:get_children_path(Path),
  N = length(L),
  lists:foreach(fun(Child) ->
    increment_count(Child) end, L),
  Counts4 = get(counts),
  OneCount = dict:fetch(?ONE, Counts4),
  ZeroCount = dict:fetch(?ZERO, Counts4),

  if
    OneCount > (N / 2) ->
      ?ONE;
    ZeroCount > (N / 2) ->
      ?ZERO;
    OneCount == ZeroCount andalso OneCount == (N / 2) ->
      strategy:get_default();
    true ->
      ?UNKNOWN
  end.


loop(Id, Config) ->
  receive
    {_From, init, AllPids} ->
      init(Id, Config, AllPids),
      loop(Id, Config);
    {_From, {receive_message, Path, Node}} ->
      receive_message(Path, Node),
      %From ! {self(), done},
      loop(Id, Config);
    {_From, {send_messages, Round, Id}} ->
      psend_messages(Round, Id),
      %From ! {self(), done},
      loop(Id, Config);
    {From, decide} ->
      From ! {self(), decide()},
      loop(Id, Config);
    {From, is_faulty} ->
      From ! {self(), strategy:is_faulty(Config, Id)},
      loop(Id, Config);
    {From, is_source} ->
      From ! {self(), config:get_source(Config) == Id},
      loop(Id, Config);
    {From, {find_majority, Path}} ->
      From ! {self(), find_majority(Path)},
      loop(Id, Config);
    Any ->
      io:format("Unexpected ~p~n", [Any]),
      loop(Id, Config)
  end.


test_process() ->
  counter:start(),
  repository:start(),
  Config = config:new(3, 5, 6),
  Pid = start(0, Config),
  init(Pid, [Pid, 1, 2, 3, 4, 5, 6]),
  Path = "mypath",
  Node = node:new(),
  receive_message(Pid, Path, Node),
  send_messages(Pid, 0, 0),
  ?UNKNOWN = decide(Pid),
  ?ONE = find_majority(Pid, Path),
  false = is_faulty(Pid),
  false = is_source(Pid).

-module(repository).
-export([start/0, reset/0, get_rank_list/2, get_children_path/1, generate_children/1, die/0, test_generate/0, test_path_ranks/0, test_children/0]).
-include("byz_general_test.hrl").


start() ->
  register(repository, spawn(fun() -> init_loop() end)).


init_loop() ->
  Children = ets:new(children, [set]),
  PathsByRank = ets:new(path_ranks, [set]),
  loop(Children, PathsByRank).

get_children_path(Path) ->
  lib_util:rpc(repository, {get_children_path, Path}).

get_rank_list(Rank, Source) ->
  lib_util:rpc(repository, {get_rank_list, Rank, Source}).

generate_children(Config) ->
  repository ! {self(), generate_children, Config}.

reset() ->
  repository ! {self(), reset}.

die() ->
  repository ! {self(), die, "Exiting"}.


table_lookup(Tab, Key) ->
  Result = ets:lookup(Tab, Key),
  case Result of
    [] ->
      [];
    error ->
      [];
    {ok, Value} ->
      lists:reverse(Value);
    [{Key, Value}] ->
      lists:reverse(Value)
  end.

table_insert(Tab, Key, Value) ->
  NewValue = case table_lookup(Tab, Key) of
               [] ->
                 [Value];
               L ->
                 [Value | L]
             end,
  ets:insert(Tab, {Key, NewValue}).



set_children_path(Children, PathKey, PathValue) ->
  table_insert(Children, PathKey, PathValue).

get_children_path(Children, Path) ->
  table_lookup(Children, Path).

get_rank_list(PathsByRank, Rank, Source) ->
  table_lookup(PathsByRank, {Rank, Source}).


set_rank_list(PathsByRank, Rank, Source, Path) ->
  Key = {Rank, Source},
  table_insert(PathsByRank, Key, Path).


generate_children(Config, Children, PathsByRank) ->
  generate_children(
    Config, Children, PathsByRank,
    config:get_source(Config),
    [],
    "",
    0).

generate_children(Config, Children, PathsByRank, Source, Ids, Path, Rank) ->
  Ids1 = [Source | Ids],
  Path1 = Path ++ integer_to_list(Source),
  set_rank_list(PathsByRank, Rank, Source, Path1),
  %%%
  Rounds = config:get_num_rounds(Config),
  if
    Rank < Rounds ->
      IdsToProcess = lists:seq(0, config:get_num_procs(Config) - 1) -- Ids1,
      lists:foreach(
        fun(Source1) -> generate_children(Config, Children, PathsByRank, Source1, Ids1, Path1, Rank + 1),
          set_children_path(Children, Path1, Path1 ++ integer_to_list(Source1))
        end, IdsToProcess);
    true ->
      true
  end,
  Children.


loop(Children, PathsByRank) ->
  receive
    {From, {get_rank_list, Rank, Source}} ->
      From ! {repository, get_rank_list(PathsByRank, Rank, Source)},
      loop(Children, PathsByRank);
    {From, {get_children_path, Path}} ->
      From ! {repository, get_children_path(Children, Path)},
      loop(Children, PathsByRank);
    {_From, reset} ->
      ets:delete_all_objects(Children),
      ets:delete_all_objects(PathsByRank),
      loop(Children, PathsByRank);
    {_From, generate_children, Config} ->
      generate_children(Config, Children, PathsByRank),
      loop(Children, PathsByRank);
    {_From, die, Reason} ->
      ets:delete(Children),
      ets:delete(PathsByRank),
      exit(Reason);
    Any ->
      io:format("Unexpected message~p~n", [Any])
  end.




benchmark_byz_general() ->
  engine:init(),
  benchmark_byz_general(5).

benchmark_byz_general(N) when N < 100 ->
  benchmark_byz_general(N, 10),
  benchmark_byz_general(N + 10);
benchmark_byz_general(N) when N >= 100 ->
  true.

benchmark_byz_general(N, M) when M < 100 ->
  statistics(runtime),
  statistics(wall_clock),
  Source = round(N / 3),
  io:format("Starting|~p|~p|~p~n", [M, N, Source]),
  engine:start(Source, N, M),
  engine:run(),
  {_, RT} = statistics(runtime),
  {_, WC} = statistics(wall_clock),
  io:format("Finished|~p|~p|~p|~p|~p|~p~n", [M, N, Source, counter:get_value(), RT, WC]),
  engine:reset(),
  benchmark_byz_general(N, M + 10);
benchmark_byz_general(_N, M) when M >= 100 ->
  true.

class Stats
     attr_accessor :processes
     attr_accessor :rounds
     attr_accessor :messages
     attr_accessor :java_time
     attr_accessor :erlang_time
     attr_accessor :cpp_time
     attr_accessor :ruby_time
     def initialize(procs, rounds, msgs)
      @processes = procs.to_i
      @rounds = rounds.to_i
      @messages = msgs.to_i
    end
    def key
      "#{@processes}/#{@rounds}"
    end
    def to_s
      "#{@processes},#{rounds},#{@messages},#{@cpp_time},#{@ruby_time},#{@java_time},#{@erlang_time}"
    end
  end
 
  stats = {}
 
  files = ["cpp.out", "ruby.out", "java.out", "erlang.out"]
  count_setter = ["cpp_time=", "ruby_time=", "java_time=", "erlang_time="]
  for i in 0...files.length
      File.open(files[i], "r").readlines.each do |line|
          if line =~ /Finish/
              t = line.split("|")
              stat = Stats.new(t[2], t[1], t[4])
              time = t[5].to_i
              stat.send(count_setter[i], time)
              stats[stat.key] = stat
           end
       end
  end
 
 
  puts "Processes,Rounds,Messages,CPP Time,Ruby Time,Java Time,Erlang Time"
  stats.values.sort_by {|stat| stat.processes * stat.rounds}.each do |stat|
    puts stat
  end