I came across an interesting blog on classical distributed Byzantine Generals Problem from Mark Nelsons’ blog. He showed C++ implementation of Leslie Lamport algorithm. It seemed like a natural fit Erlang, but instead of writing directly in Erlang, I first wrote the algorithm into Java and Ruby (with slight redesign). Unfortunately, the original C++ source, and my Java and Ruby versions are not really truly distributed or concurrent. I just wanted to translate the original algorithm without changing a whole a lot, but Erlang gave you both distributing features and concurrency for free. You can learn about the Byzantine General problem from Mark Nelson’s post or from this brief summary, or the Wikipedia. The algorithm is heavily based on message communication where a general sends a message value (0 or 1), to his lieutenants and each lieutenant sends the value to all other lieutenants including himself. After some specific number of rounds a consensus is reached. The original C++ code defines a Node structure to store input/output values of each process, a Traits class to store configuration and a method to come up with new values. Since, some processes can be faulty (or traiters), the Traits class determines the value that a process will return. The most of the fun happens in the Process class that performs all message communication. You can download the original C++ source code.
I slightly changed the structure and got rid of static attributes and methods. In my design, the application consists of following modules:
- Configuration – for storing configuration for simulation such as number of processes, rounds, source (general) process.
- Strategy – that determines the value that should be returned by the process.
- Repository – that stores hierarchical paths for messages that were communicated by the processes.
- Process – GeneralProcesses classification that send or receive messages.
- Engine – drives the algorithm.
- Test – that actually invokes engine with different parameters and stores timings.
Here is the Java implementation (I am showing multiple java classes here, but you can download the complete source code from here. I kept most of the algorithm as it was, which turned out to be really memory hog when you increase number of processes. In fact, I could not run the Java application for more than 10 processes with 512MB memory. Here is the source code for the Java version:
Java
x
public enum Value { ZERO, ONE, FAULTY, UNKNOWN}Java
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);}Java
public interface Broadcaster { public void broadcast(int round, int id, Node source, String path);}Java
public interface PathRepository { public void generateChildren(); public List<String> getRankList(int rank, int source); public List<String> getPathChildren(String path);}Java
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; }}Java
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(); }}Java
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)); }}Java
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; }}Java
public interface Engine extends Runnable { public void start();}Java
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()); } 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; }}Java
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; }}Here is the ruby version, which is pretty much same as the Java version (except a bit smaller). The operator overloading and power of collections give Ruby the terseness over Java version. Again, the application is consisted of classes for process, strategy, repository, engine, node, and config. The benchmarks are kicked off from ByzGeneralTest, which calls Engine, which in turn creates Repository, Strategy, Generals and LProcesses (lieutenants).
Ruby
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}" endendRuby
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}" endendRuby
class Value ZERO = '0' ONE = '1' FAULTY = 'X' UNKNOWN = '?'endRuby
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("") endendprivatedef 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.dupendendRuby
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 endendclass 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 endendpath_list = @repository.rank_list(0, @config.source)top_path = path_list[0]@nodes[top_path].outputenddef 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_defaultelse Value::UNKNOWNendenddef faulty? @strategy.faulty?(id)enddef source? @config.source == idendendclass GeneralProcess < LProcess def initialize(id, config, repository, broadcaster, strategy) super(id, config, repository, broadcaster, strategy) @nodes[""] = @strategy.source_value end def decide @nodes[""].input endendRuby
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 endenddef run for i in 0...@config.num_rounds for j in 0...@config.num_procs @procs[j].send_messages(i)endendfor j in 0...@config.num_procsprint("Source ") if @procs[j].source?print("Process #{j}")unless @procs[j].faulty? print("decides on value #{@procs[j].decide}")else print(" is faulty")endputs("")endenddef 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_childrenendendclass 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 endendTest::Unit::UI::Console::TestRunner.run(ByzGeneralTest)Finally here is the Erlang source code (Again I am showing multiple modules here). I tried to break the structure same as my Java and Ruby version and the application consisted of process, strategy, repository, engine, node, and config modules. The benchmarks were kicked off from the byz_general_test, which invoked engine and engine created processes for repository, generals and lieutenants. Another distinction is difference between active objects and regular functions. In this design, repository and processes are active objects that can receive messages via Erlang’s message communication primitives. Also, I added message counter just to keep track of number of messages for the simulation (though it is not requirement for the algorithm). I used ets APIs in repository to store paths of messages.
Erlang
-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_roundsErlang
-module(counter).-export([, , , , , ]).-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()Erlang
-module(engine).-export([, , , , ]).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.Erlang
-module(lib_util).-compile(export_all).rpc(Pid, Request) -> Pid ! {self(), Request}, receive {Pid, Response} -> Response end.Erlang
-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}.Erlang
-module(process).-export([, , , , , , , , ]).-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(),ifSource =:= Id ->Nodes1 = dict:store("", SourceValue, Nodes),put(nodes, Nodes1);true ->trueend .%%%% 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).Erlang
-module(repository).-export([, , , , , , , , ]).-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.I merged the results with following script:
Ruby
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 And here are the results of benchmark that I ran on my Dell notebook (Note I used time function in C++ which only returns timing in seconds, so C++ timings are actually not precise.):
Conclusion:
This was interesting problem that tested limits of these languages. For example, I found when using more than 10 processes things got really nasty. The Java program gave up with OutOfMemory. The Erlang program dumped crashed, though I would have used OTP’s supervisors if I was writing more fault tolerant application. Ruby program became too slow, so I had to kill it. Java turned out to be the performance leader in this case and I was a bit surprised when Erlang’s response time became really high with 9 processes. As, I mentioned earlier, the C++, Java and Ruby versions are not really concurrent and their message passing is really method invocation. As far as mapping algorithm to the language, I found that Erlang fit very nicely for distributed algorithms that involve a lot of communications. I left the C++, Java and Ruby version very simple and didn’t try to implement truly independent processes and communication because that would have required a lot more effort.