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:
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;
}
} 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).
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
endclass 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
endclass 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
endclass 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
endclass 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)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.
-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.I merged the results with following script:
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.
Resources:
Source Code