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.
First, 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 the 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 consenus 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 – that 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. Also, in my Java versio I used Interface Separation Principle (old habit):
1: 2: public enum Value { 3: ZERO, 4: ONE, 5: FAULTY, 6: UNKNOWN 7: } 8: public interface ValueStrategy { 9: 10: public Node getSourceValue(); 11: 12: public Value getValue( 13: Value value, 14: int source, 15: int destination, 16: String path); 17: 18: 19: public Value getDefaultValue(); 20: 21: 22: public boolean isFaulty(int process); 23: } 24: 25: 26: 27: public interface Broadcaster { 28: public void broadcast(int round, int id, Node source, String path); 29: } 30: 31: 32: 33: 34: import java.util.*; 35: 36: public interface PathRepository { 37: 38: public void generateChildren(); 39: 40: public List<String> getRankList(int rank, int source); 41: 42: public List<String> getPathChildren(String path); 43: } 44: 45: 46: 47: public class Configuration { 48: private int source; 49: private int roundsOfMessages; 50: private int numberOfProcesses; 51: 52: // 53: Configuration(int source, int roundsOfMessages, int numberOfProcesses) { 54: this.source = source; 55: this.roundsOfMessages = roundsOfMessages; 56: this.numberOfProcesses = numberOfProcesses; 57: } 58: 59: // 60: public int getSource() { 61: return source; 62: } 63: 64: 65: public int getRoundsOfMessages() { 66: return roundsOfMessages; 67: } 68: 69: 70: public int getNumberOfProcesses(){ 71: return numberOfProcesses; 72: } 73: } 74: 75: 76: 77: 78: import java.util.*; 79: 80: public class DefaultEngine implements Engine, Broadcaster { 81: static final boolean debug = System.getProperty("debug", "false").equals("true"); 82: private Configuration configuration; 83: private PathRepository repository; 84: private List<Process> processes; 85: private ValueStrategy strategy; 86: 87: // 88: public DefaultEngine(int source, int roundsOfMessages, int numberOfProcesses) { 89: this.configuration = new Configuration(source, roundsOfMessages, numberOfProcesses); 90: this.strategy = new DefaultValueStrategy(configuration); 91: this.repository = new DefaultPathRepository(configuration); 92: this.processes = new ArrayList<Process>(); 93: } 94: 95: public void broadcast(int round, int id, Node source, String path) { 96: for (int j=0; j<configuration.getNumberOfProcesses(); j++) { 97: if ( j != configuration.getSource()) { 98: Value value = strategy.getValue( 99: source.input, 100: id, 101: j, 102: path); 103: if (debug) { 104: String sourcePath = path.substring(0, path.length()-1); 105: System.out.println("Sending for round " + round + " from process " + id + " to " + j + ": {" + value + 106: ", " + path + ", " + Value.UNKNOWN + "}, getting value from source " + sourcePath); 107: } 108: getProcesses().get(j).receiveMessage(path, new Node(value, Value.UNKNOWN)); 109: } 110: } 111: } 112: 113: 114: public List<Process> getProcesses() { 115: return processes; 116: } 117: 118: public void run() { 119: // 120: // Starting at round 0 and working up to round M, call the 121: // SendMessages() method of each process. It will send the appropriate 122: // message to all other sibling processes. 123: // 124: for (int i=0; i<= configuration.getRoundsOfMessages(); i++) { 125: for (int j=0; j<configuration.getNumberOfProcesses(); j++) { 126: processes.get(j).sendMessages(i); 127: } 128: } 129: 130: // 131: // All that is left is to print out the results. For non-faulty processes, 132: // we call the Decide() method to see what what the process decision was 133: // 134: for (int j=0; j<configuration.getNumberOfProcesses(); j++) { 135: if (processes.get(j).isSource()) System.out.print("Source "); 136: System.out.print("Process " + j); 137: if (!processes.get(j).isFaulty()) { 138: System.out.print(" decides on value " + 139: processes.get(j).decide()); 140: } else { 141: System.out.print(" is faulty"); 142: } 143: System.out.println(); 144: } 145: } 146: 147: 148: public void start() { 149: // 150: for (int i=0; i<configuration.getNumberOfProcesses(); i++) { 151: Process process = i == configuration.getSource() ? new GeneralProcess(i, configuration, repository, this, strategy) : new Process(i, configuration, repository, this, strategy); 152: processes.add(process); 153: } 154: repository.generateChildren(); 155: } 156: } 157: import java.util.*; 158: 159: public class DefaultPathRepository implements PathRepository { 160: private Map<String, List<String>> children; 161: private Map<Integer, Map<Integer, List<String>>> pathByRank; 162: private Configuration configuration; 163: static final boolean debug = System.getProperty("debug", "false").equals("true"); 164: 165: // 166: public DefaultPathRepository(Configuration configuration) { 167: this.configuration = configuration; 168: this.children = new HashMap<String, List<String>>(); 169: this.pathByRank = new HashMap<Integer, Map<Integer, List<String>>>(); 170: } 171: 172: public Map<String, List<String>> getChildren() { 173: return children; 174: } 175: 176: public List<String> getPathChildren(String path) { 177: List<String> pathList = children.get(path); 178: if (pathList == null) { 179: pathList = new ArrayList<String>(); 180: children.put(path, pathList); 181: } 182: return pathList; 183: } 184: 185: 186: public List<String> getRankList(int rank, int source) { 187: Map<Integer, List<String>> pathMap = pathByRank.get(rank); 188: if (pathMap == null) { 189: pathMap = new HashMap<Integer, List<String>>(); 190: pathByRank.put(rank, pathMap); 191: } 192: List<String> pathList = pathMap.get(source); 193: if (pathList == null) { 194: pathList = new ArrayList<String>(); 195: pathMap.put(source, pathList); 196: } 197: return pathList; 198: } 199: 200: public void generateChildren() { 201: generateChildren(configuration.getSource(), new boolean[configuration.getNumberOfProcesses()], "", 0); 202: } 203: 204: 205: private void generateChildren(int source, boolean[] ids, String path, int rank) { 206: ids[source] = true; 207: path += toCharString(source); 208: getRankList(rank, source).add(path); 209: 210: // 211: if (rank < configuration.getRoundsOfMessages()) { 212: for (int i=0; i<ids.length; i++) { 213: if (!ids[i]) { 214: boolean[] newIds = new boolean[ids.length]; 215: System.arraycopy(ids, 0, newIds, 0, ids.length); 216: generateChildren(i, newIds, path, rank + 1); 217: getPathChildren(path).add(path + toCharString(i)); 218: } 219: } 220: } 221: 222: // 223: if (debug) { 224: System.out.print("generateChildren(" + source + "," + rank + "," + path + "), children = "); 225: List<String> list = getPathChildren(path); 226: for (String s : list) { 227: System.out.print(s + " "); 228: } 229: System.out.println(); 230: } 231: } 232: 233: 234: // 235: private char toChar(int n) { 236: return (char) (n + '0'); 237: } 238: private String toCharString(int n) { 239: return String.valueOf(toChar(n)); 240: } 241: } 242: 243: 244: 245: 246: 247: 248: import java.util.*; 249: 250: public class DefaultValueStrategy implements ValueStrategy { 251: private Configuration configuration; 252: private Node sourceValue; 253: 254: // 255: DefaultValueStrategy(Configuration configuration) { 256: this.configuration = configuration; 257: this.sourceValue = new Node(Value.ZERO, Value.UNKNOWN); 258: } 259: 260: // 261: // This method returns the true value of the source's value. The source may send 262: // faulty values to other processes, but the Node returned by this value will be 263: // in its root node. 264: // 265: // In this case, the General's node has in input value of 0, which makes that 266: // the desired value. Of course, since the General is faulty, this doesn't really 267: // matter. 268: // 269: public Node getSourceValue() { 270: return sourceValue; 271: } 272: 273: // 274: // During message, GetValue() is called to get the value returned by a given process 275: // during a messaging round. 'value' is the input value that it should be sending to 276: // the destination process (if it isn't faulty), source is the source process ID, 277: // destination is the destination process ID, and Path is the path being used for this 278: // particular message. 279: // 280: // In this particular implementation, we have two faulty processes - the source 281: // process, which returns a sort-of random value, and process ID 2, which returns 282: // a ONE always, in contradiction of the General's desired value of 0. 283: // 284: public Value getValue(Value value, int source, int destination, String path) { 285: if (configuration.getSource() == source) return (destination & 1) != 0 ? Value.ZERO : Value.ONE; 286: else if (source == 2) return Value.ONE; 287: return value; 288: } 289: 290: 291: // 292: // When breaking a tie, GetDefault() is used to get the default value. 293: // 294: // This is an arbitrary decision, but it has to be consistent across all processes. 295: // More importantly, the processes have to arrive at a correct decision regardless 296: // of whether the default value is always 0 or always 1. In this case we've set it to 297: // a value of 1. 298: // 299: public Value getDefaultValue() { 300: return Value.ONE; 301: } 302: 303: // 304: // This method is used to identify fault processes by ID 305: // 306: public boolean isFaulty(int process) { 307: return process == configuration.getSource() || process == 2; 308: } 309: } 310: 311: public interface Engine extends Runnable { 312: public void start(); 313: } 314: public class GeneralProcess extends Process { 315: public GeneralProcess(int id, Configuration configuration, PathRepository repository, Broadcaster broadcaster, ValueStrategy strategy) { 316: super(id, configuration, repository, broadcaster, strategy); 317: nodes.put("", strategy.getSourceValue()); 318: } 319: 320: 321: @Override 322: public Value decide() { 323: // 324: // The source process doesn't have to do all the work - since it's the decider, 325: // it simply looks at its input value to pick the appropriate decision value. 326: // 327: return nodes.get("").input; 328: } 329: } 330: 331: public class Node { 332: Value input; 333: Value output; 334: Node() { 335: this(Value.FAULTY, Value.FAULTY); 336: } 337: Node(Value input, Value output) { 338: this.input = input; 339: this.output = output; 340: } 341: } 342: 343: 344: import java.util.*; 345: 346: 347: public class Process { 348: protected int id; 349: protected Configuration configuration; 350: protected ValueStrategy strategy; 351: protected Map<String, Node> nodes; 352: protected PathRepository repository; 353: protected Broadcaster broadcaster; 354: static final boolean debug = System.getProperty("debug", "false").equals("true"); 355: public Process(int id, Configuration configuration, PathRepository repository, Broadcaster broadcaster, ValueStrategy strategy) { 356: this.id = id; 357: this.strategy = strategy; 358: this.repository = repository; 359: this.broadcaster = broadcaster; 360: this.configuration = configuration; 361: this.nodes = new HashMap<String, Node>(); 362: } 363: static int totalMessages; 364: // 365: // 366: // Receiving a message is pretty simple here, it means that some other process 367: // calls this method on the current process with path and a node. All we do 368: // is store the value, we'll use it in the next round of messaging. 369: // 370: public void receiveMessage(String path, Node node) { 371: nodes.put(path, node); 372: totalMessages++; 373: } 374: 375: 376: // 377: // After constructing all messages, you need to call SendMessages on each process, 378: // once per round. This routine will send the appropriate messages for each round 379: // to all th eother processes listed in the vector passed in as an argument. 380: // 381: // Deciding on what messages to send is pretty simple. If we look at the static 382: // map pathsByRank, indexed by round and the processId of this process, it gives 383: // the entire set of taraget paths that this process needs to send messages to. 384: // So there is an iteration loop through that map, and this process sends a message 385: // to the correct target process for each path in the map. 386: // 387: public void sendMessages(int round) { 388: List<String> pathList = repository.getRankList(round, id); 389: for (String path : pathList) { 390: String sourcePath = path.substring(0, path.length()-1); 391: Node source = nodes.get(sourcePath); 392: if (source == null) throw new IllegalStateException("Failed to find source node for " + sourcePath); 393: broadcaster.broadcast(round, id, source, path); 394: } 395: } 396: 397: // 398: // After all messages have been sent, it's time to Decide. 399: // 400: // This part of the algorithm follows the description in the article closely. 401: // It has to work its way from the leaf values up to the root of the tree. 402: // The first round consists of going to the leaf nodes, and copying the input 403: // value to the output value. 404: // 405: // All subsequent rounds consist of getting the majority of the output values from 406: // each nodes children, and copying that to the nodes output value. 407: // 408: // When we finally reach the root node, there is only one node with an output value, 409: // and that represents this processes decision. 410: // 411: public Value decide() { 412: // 413: // Step 1 - set the leaf values 414: // 415: for (int i=0; i<configuration.getNumberOfProcesses(); i++) { 416: List<String> pathList = repository.getRankList(configuration.getRoundsOfMessages(), i); 417: for (String path : pathList) { 418: Node node = nodes.get(path); 419: node.output = node.input; 420: } 421: } 422: // 423: // Step 2 - work up the tree 424: // 425: for (int round = configuration.getRoundsOfMessages() - 1 ; round >= 0 ; round--) { 426: for (int i=0; i<configuration.getNumberOfProcesses(); i++) { 427: List<String> pathList = repository.getRankList(round, i); 428: for (String path : pathList) { 429: Node node = nodes.get(path); 430: node.output = getMajority(path); 431: } 432: } 433: } 434: List<String> pathList = repository.getRankList(0, configuration.getSource()); 435: String topPath = pathList.get(0); 436: return nodes.get(topPath).output; 437: } 438: 439: // 440: // This routine calculates the majority value for the children of a given 441: // path. The logic is pretty simple, we increment the count for all possible 442: // values over the children. If there is a clearcut majority, we return that, 443: // otherwise we return the default value defined by the strategy class. 444: // 445: public Value getMajority(String path) { 446: Map<Value, Integer> counts = new HashMap<Value, Integer>(); 447: counts.put(Value.ONE, new Integer(0)); 448: counts.put(Value.ZERO, new Integer(0)); 449: counts.put(Value.UNKNOWN, new Integer(0)); 450: Collection<String> list = repository.getPathChildren(path); 451: int n = 0; 452: if (list == null) { 453: if (debug) System.out.println("No child found for '" + path + "'"); 454: } else { 455: n = list.size(); 456: for (String child : list) { 457: Node node = nodes.get(child); 458: if (node != null) { 459: counts.put(node.output, counts.get(node.output) + 1); 460: } else if (debug) { 461: //System.out.println("Could not find node for count with path " + child); 462: } 463: } 464: } 465: // 466: // 467: if (counts.get(Value.ONE) > ( n / 2 ) ) return Value.ONE; 468: else if (counts.get(Value.ZERO) > ( n / 2 ) ) return Value.ZERO; 469: else if (counts.get(Value.ONE).intValue() == counts.get(Value.ZERO).intValue() && 470: counts.get(Value.ONE) == (n / 2)) return strategy.getDefaultValue(); 471: return Value.UNKNOWN; 472: } 473: 474: 475: // 476: // A utility routine that tells whether a given process is faulty 477: // 478: public boolean isFaulty() { 479: return strategy.isFaulty(id); 480: } 481: // 482: // Another somewhat handy utility routine 483: // 484: public boolean isSource() { 485: return configuration.getSource() == id; 486: } 487: } 488: 489:
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).
1: require 'benchmark' 2: require 'test/unit' 3: require 'test/unit/ui/console/testrunner' 4: 5: DEBUG = false 6: 7: 8: 9: 10: class Configuration 11: attr_reader :source, :num_rounds, :num_procs 12: def initialize(source, num_rounds, num_procs) 13: @source = source 14: @num_rounds = num_rounds 15: @num_procs = num_procs 16: end 17: def to_s 18: "#{@source}|#{@num_rounds}|#{@num_procs}" 19: end 20: end 21: 22: class Node 23: attr_accessor :input 24: attr_accessor :output 25: def initialize(input = Value::FAULTY, output = Value::FAULTY) 26: @input = input 27: @output = output 28: end 29: def to_s 30: "#{@input}/#{@output}" 31: end 32: end 33: 34: class Value 35: ZERO = '0' 36: ONE = '1' 37: FAULTY = 'X' 38: UNKNOWN = '?' 39: end 40: 41: 42: 43: 44: 45: class PathRepository 46: attr_accessor :children 47: attr_accessor :path_by_rank 48: attr_accessor :config 49: 50: def initialize(config) 51: @config = config 52: @children = Hash.new {|h,k| h[k] = []} 53: @path_by_rank = Hash.new {|h,k| h[k] = Hash.new {|hh,kk| hh[kk] = []}} 54: end 55: 56: def path_children(path) 57: @children[path] 58: end 59: 60: def rank_list(rank, source) 61: @path_by_rank[rank][source] 62: end 63: 64: def generate_children(source = @config.source, ids = new_ids, path = "", rank = 0) 65: ids[source] = true 66: path = "#{path}#{source}" 67: @path_by_rank[rank][source].push path 68: 69: if rank < @config.num_rounds 70: for i in 0...@config.num_procs 71: unless ids[i] 72: generate_children(i, new_ids(ids), path, rank + 1) 73: @children[path].push("#{path}#{i}") 74: end 75: end 76: end 77: 78: if (DEBUG) 79: print("generate_children(#{source}, #{rank}, #{path}, children = ") 80: @children[path].each do |child| 81: print(child + " ") 82: end 83: puts("") 84: end 85: end 86: 87: private 88: def new_ids(old_ids = nil) 89: #old_ids.nil? ? Array.new(@config.num_procs) {|i| false} : Array.new(old_ids) 90: old_ids.nil? ? Hash.new {|h,k| h[k] = false} : old_ids.dup 91: end 92: end 93: 94: 95: 96: class ValueStrategy 97: attr_accessor :config 98: attr_accessor :source_value 99: 100: def initialize(config) 101: @config = config 102: @source_value = Node.new(Value::ZERO, Value::UNKNOWN) 103: end 104: 105: def create_value(value, source, destination, path) 106: if @config.source == source 107: (destination & 1) != 0 ? Value::ZERO : Value::ONE 108: elsif source == 2 109: Value::ONE 110: else 111: value 112: end 113: end 114: 115: 116: def get_default 117: return Value::ONE 118: end 119: 120: def faulty?(process) 121: process == @config.source || process == 2 122: end 123: end 124: 125: 126: 127: class LProcess 128: @@total_messages = 0 129: attr_reader :id 130: attr_reader :config 131: attr_reader :strategy 132: attr_reader :repository 133: attr_reader :broadcaster 134: # 135: def initialize(id, config, repository, broadcaster, strategy) 136: @id = id 137: @config = config 138: @repository = repository 139: @strategy = strategy 140: @broadcaster = broadcaster 141: @nodes = Hash.new 142: end 143: 144: 145: def self.total_messages 146: @@total_messages 147: end 148: def self.reset_total_messages 149: @@total_messages = 0 150: end 151: 152: def receive_message(path, node) 153: @nodes[path] = node 154: @@total_messages += 1 155: end 156: 157: def send_messages(round) 158: @repository.rank_list(round, @id).each do |path| 159: source_path = path.slice(0, path.length-1) 160: source = @nodes[source_path] 161: raise "Source path #{source_path} not found" if source.nil? 162: broadcaster.broadcast(round, @id, source, path) 163: end 164: end 165: 166: def decide 167: #### Step 1 - set the leaf values 168: for i in 0...@config.num_procs 169: @repository.rank_list(@config.num_rounds, i).each do |path| 170: node = @nodes[path] 171: node.output = node.input 172: end 173: end 174: 175: ### Step 2 - work up the tree 176: (@config.num_rounds - 1).step 0, -1 do |round| 177: for i in 0...@config.num_procs 178: @repository.rank_list(round, i).each do |path| 179: @nodes[path].output = find_majority(path) 180: end 181: end 182: end 183: 184: path_list = @repository.rank_list(0, @config.source) 185: top_path = path_list[0] 186: @nodes[top_path].output 187: end 188: 189: def find_majority(path) 190: counts = Hash.new(0) 191: list = @repository.path_children(path) 192: n = 0 193: if list 194: n = list.size 195: list.each do |child| 196: node = @nodes[child] 197: counts[node.output] = counts[node.output] + 1 if node 198: end 199: end 200: # 201: if (counts[Value::ONE] > ( n / 2 ) ) 202: Value::ONE 203: elsif (counts[Value::ZERO] > ( n / 2 ) ) 204: Value::ZERO 205: elsif (counts[Value::ONE] == counts[Value::ZERO] && 206: counts[Value::ONE] == (n / 2)) 207: @strategy.get_default 208: else 209: Value::UNKNOWN 210: end 211: end 212: 213: def faulty? 214: @strategy.faulty?(id) 215: end 216: 217: def source? 218: @config.source == id 219: end 220: end 221: 222: 223: class GeneralProcess < LProcess 224: def initialize(id, config, repository, broadcaster, strategy) 225: super(id, config, repository, broadcaster, strategy) 226: @nodes[""] = @strategy.source_value 227: end 228: 229: 230: def decide 231: @nodes[""].input 232: end 233: end 234: 235: class Engine 236: attr_reader :config 237: attr_reader :repository 238: attr_reader :procs 239: attr_reader :strategy 240: 241: def initialize(source, num_rounds, num_procs) 242: @config = Configuration.new(source, num_rounds, num_procs) 243: @strategy = ValueStrategy.new(config) 244: @repository = PathRepository.new(config) 245: @procs = [] 246: end 247: 248: def broadcast(round, id, source, path) 249: for j in 0...@config.num_procs 250: unless j == @config.source 251: value = @strategy.create_value(source.input, id, j, path) 252: if (DEBUG) 253: source_path = path.slice(0, path.length-1) 254: puts("Sending for round #{round} from process #{id} to #{j} : #{value}, #{path}, #{Value::UNKNOWN}, getting value from source #{source_path}") 255: end 256: @procs[j].receive_message(path, Node.new(value, Value::UNKNOWN)) 257: end 258: end 259: end 260: 261: 262: def run 263: for i in 0...@config.num_rounds 264: for j in 0...@config.num_procs 265: @procs[j].send_messages(i) 266: end 267: end 268: 269: for j in 0...@config.num_procs 270: print("Source ") if @procs[j].source? 271: print("Process #{j}") 272: unless @procs[j].faulty? 273: print("decides on value #{@procs[j].decide}") 274: else 275: print(" is faulty") 276: end 277: puts("") 278: end 279: end 280: 281: 282: def start 283: for i in 0...@config.num_procs 284: process = i == @config.source ? 285: GeneralProcess.new(i, @config, @repository, self, @strategy) : 286: LProcess.new(i, @config, @repository, self, @strategy) 287: @procs.push(process) 288: end 289: @repository.generate_children 290: end 291: end 292: 293: 294: class ByzGeneralTest < Test::Unit::TestCase 295: def setup 296: end 297: 298: 299: def run_engine(m, n, source) 300: r = Benchmark.realtime() do 301: puts("Starting|#{m}|#{n}|#{source}") 302: engine = Engine.new(source, m, n) 303: engine.start 304: engine.run 305: end 306: puts("Finished|#{m}|#{n}|#{source}|#{LProcess.total_messages}|#{r*1000}") 307: Process.reset_total_messages 308: end 309: 310: 311: def xtest_once 312: n = 5 313: m = 6 314: source = 3 315: run_engine(m, n, source) 316: end 317: 318: def test_multiple 319: 5.step 50, 5 do |n| 320: 10.step 100, 10 do |m| 321: source = n / 3 322: run_engine(m, n, source) 323: end 324: end 325: end 326: end 327: Test::Unit::UI::Console::TestRunner.run(ByzGeneralTest) 328:
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.
1: 2: 3: -module(config). 4: -compile(export_all). 5: 6: 10: -record(config, {source, num_procs, num_rounds}). 11: 12: new(S, N, M) -> 13: #config{source = S, num_procs = N, num_rounds = M}. 14: 15: get_source(Self) -> 16: Self#config.source. 17: get_num_procs(Self) -> 18: Self#config.num_procs. 19: get_num_rounds(Self) -> 20: Self#config.num_rounds. 21: 25: -module(counter). 26: -export([start/0, get_value/0, increment/0, reset/0, die/0, test_counter/0]). 27: -include("byz_general_test.hrl"). 28: 29: 33: 34: 35: start() -> 36: register(message_counter, spawn(fun() -> loop(0) end)). 37: 38: 39: get_value() -> 40: lib_util:rpc(message_counter, value). 41: 42: 43: increment() -> 44: message_counter ! {self(), increment}. 45: 46: reset() -> 47: message_counter ! {self(), reset}. 48: 49: 50: die() -> 51: message_counter ! {self(), die, "Exiting"}. 52: 53: 54: 55: loop(N) -> 56: receive 57: {_From, increment} -> 58: loop(N+1); 59: {_From, reset} -> 60: loop(0); 61: {From, value} -> 62: From ! {message_counter, N}, 63: loop(N); 64: {_From, die, Reason} -> 65: unregister(message_counter), 66: exit(Reason); 67: Any -> 68: io:format("Unexpected message ~p~n", [Any]) 69: end. 70: 71: 72: 76: test_counter() -> 77: start(), 78: increment(), 79: increment(), 80: 2 = get_value(), 81: reset(), 82: 0 = get_value(), 83: die(). 87: -module(engine). 88: -export([start/3, init/0, run/0, reset/0, test/0]). 89: 90: 94: 95: init() -> 96: counter:start(), 97: repository:start(). 98: 99: reset() -> 100: counter:reset(), 101: repository:reset(). 102: 103: start(Source, N, M) -> 104: Config = config:new(Source, N, M), 105: put(config, Config), 106: ProcIds = lists:seq(0, config:get_num_procs(Config)-1), 107: Pids = lists:map(fun(I) -> process:start(I, Config) end, ProcIds), 108: put(pids, Pids), 109: lists:foreach(fun(Pid) -> process:init(Pid, Pids) end, Pids), 110: repository:generate_children(Config), 111: Config. 112: 113: run() -> 114: Config = get(config), 115: Ids = lists:seq(0, config:get_num_procs(Config)-1), 116: lists:foreach(fun(Id) -> send_message(Id) end, Ids), 117: lists:foreach(fun(Id) -> print_result(Id) end, Ids). 118: 122: get_pid(Id) -> 123: Pids = get(pids), 124: lists:nth(Id+1, Pids). 125: 126: send_message(Id) -> 127: Config = get(config), 128: RoundIds = lists:seq(0, config:get_num_rounds(Config)-1), 129: lists:foreach(fun(Round) -> send_message(Id, Round) end, RoundIds). 130: 131: send_message(Id, Round) -> 132: Pid = get_pid(Id), 133: process:send_messages(Pid, Round, Id). 134: 135: print_result(Id) -> 136: Config = get(config), 137: Source = config:get_source(Config), 138: case Source of 139: Id -> 140: io:format("Source "); 141: _ -> 142: true 143: end, 144: io:format("Process ~p ", [Id]), 145: 146: Pid = get_pid(Id), 147: Faulty = process:is_faulty(Pid), 148: if 149: Faulty -> 150: io:format(" is faulty~n"); 151: true -> 152: Decision = process:decide(Pid), 153: io:format(" decides on value ~p~n", [Decision]) 154: end. 155: 159: test() -> 160: init(), 161: Config = start(3, 5, 6), 162: run(), 163: Config. 164: -module(lib_util). 165: -compile(export_all). 166: 170: rpc(Pid, Request) -> 171: Pid ! {self(), Request}, 172: receive 173: {Pid, Response} -> 174: Response 175: end. 176: -module(node). 177: -include("byz_general_test.hrl"). 178: -compile(export_all). 179: 183: -record(node, {input, output}). 184: 185: new() -> 186: #node{input = ?FAULTY, output = ?FAULTY}. 187: new(I, O) -> 188: #node{input = I, output = O}. 189: 190: 191: set_input(Self, I) -> 192: Self#node{input = I}. 193: get_input(Self) -> 194: Self#node.input. 195: 196: set_output(Self, O) -> 197: Self#node{output = O}. 198: get_output(Self) -> 199: Self#node.output. 200: set_output_as_input(Self) -> 201: O = get_output(Self), 202: Self#node{output = O}. 203: 204: 205: 206: 207: 211: -module(process). 212: -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]). 213: -include("byz_general_test.hrl"). 214: 215: 220: 221: % 222: %%% starts process loop 223: % 224: start(Id, Config) -> 225: spawn(fun() -> loop(Id, Config) end). 226: 227: receive_message(Pid, Path, Node) -> 228: %lib_util:rpc(Pid, {receive_message, Path, Node}). 229: Pid ! {self(), {receive_message, Path, Node}}. 230: 231: send_messages(Pid, Round, Id) -> 232: %lib_util:rpc(Pid, {send_messages, Round, Id}). 233: Pid ! {self(), {send_messages, Round, Id}}. 234: 235: init(Pid,Pids) -> 236: Pid ! {self(), init, Pids}. 237: 238: decide(Pid) -> 239: lib_util:rpc(Pid, decide). 240: 241: find_majority(Pid, Path) -> 242: lib_util:rpc(Pid, {find_majority, Path}). 243: 244: is_faulty(Pid) -> 245: lib_util:rpc(Pid, is_faulty). 246: 247: is_source(Pid) -> 248: lib_util:rpc(Pid, is_source). 249: 254: put(config, Config), 255: put(allPids, AllPids), 256: SourcePid = lists:nth(config:get_source(Config)+1, AllPids), 257: put(lieutenants, AllPids -- [SourcePid]), 258: Nodes = dict:new(), 259: put(nodes, Nodes), 260: Source = config:get_source(Config), 261: SourceValue = strategy:source_value(), 262: if 263: Source =:= Id -> 264: Nodes1 = dict:store("", SourceValue, Nodes), 265: put(nodes, Nodes1); 266: true -> 267: true 268: end. 269: 270: 271: 272: % 273: %%% stores message receievd in nodes dictionary 274: %%% we are also keeping track of total number of messages received in simulation. 275: % 276: receive_message(Path, Node) -> 277: Nodes = get(nodes), 278: Nodes1 = dict:store(Path, Node, Nodes), 279: put(nodes, Nodes1), 280: counter:increment(). 281: 282: get_node(Path) -> 283: Nodes = get(nodes), 284: Response = dict:find(Path, Nodes), 285: case Response of 286: {ok, Node} -> 287: Node; 288: _ -> 289: %io:format("Failed to find node for path ~p~n", [Path]), 290: node:new() 291: end. 292: 293: % 294: %%% 295: % 296: psend_messages(Round, Id) -> 297: L = repository:get_rank_list(Round, Id), 298: lists:foreach( 299: fun(Path) -> 300: psend_messages(Round, Id, Path) end, L). 301: 302: psend_messages(Round, Id, Path) -> 303: Length = string:len(Path), 304: SourcePath = if 305: Length > 1 -> 306: string:substr(Path, 1, Length-1); 307: true -> 308: "" 309: end, 310: Source = get_node(SourcePath), 311: broadcast(Round, Id, Source, Path). 312: 313: 314: % 315: %%% broadcast message to all processes 316: % 317: broadcast(Round, Id, SourceNode, Path) -> 318: Config = get(config), 319: ProcIds = lists:seq(0, config:get_num_procs(Config)-1), 320: IdsToProcess = ProcIds -- [config:get_source(Config)], 321: lists:foreach( 322: fun(Dest) -> 323: broadcast(Round, Id, SourceNode, Path, Dest) end, IdsToProcess). 324: 325: broadcast(_Round, Id, SourceNode, Path, Dest) -> 326: Config = get(config), 327: Value = strategy:create_value(Config, node:get_input(SourceNode), Id, Dest, Path), 328: ProcPids = get(allPids), 329: ProcPid = lists:nth(Dest+1, ProcPids), 330: receive_message(ProcPid, Path, node:new(Value, ?UNKNOWN)). 331: 332: 333: decide() -> 334: %%% Step 1 - set the leaf values 335: Config = get(config), 336: ProcIds = lists:seq(0, config:get_num_procs(Config)-1), 337: lists:foreach(fun(X) -> reset_node(X) end, ProcIds), 338: %%% Step 2 - work up the tree 339: RoundIds = lists:reverse(lists:seq(0, config:get_num_rounds(Config)-1)), 340: lists:foreach( 341: fun(Round) -> 342: set_node_value(Round) end, RoundIds), 343: Source = config:get_source(Config), 344: Result = repository:get_rank_list(0, Source), 345: case Result of 346: [] -> 347: ?UNKNOWN; 348: [TopPath] -> 349: node:get_output(get_node(TopPath)); 350: [TopPath|_] -> 351: node:get_output(get_node(TopPath)) 352: end. 353: 354: 355: set_node_value(Round) -> 356: Config = get(config), 357: ProcIds = lists:seq(0, config:get_num_procs(Config)-1), 358: lists:foreach( 359: fun(Id) -> 360: set_node_value(Round, Id) end, ProcIds). 361: 362: set_node_value(Round, Id) -> 363: L = repository:get_rank_list(Round, Id), 364: lists:foreach( 365: fun(Path) -> 366: set_node_value(Round, Id, Path) end, L). 367: 368: set_node_value(_Round, _Id, Path) -> 369: Nodes = get(nodes), 370: Node = get_node(Path), 371: Value = find_majority(Path), 372: Nodes1 = dict:store(Path, node:set_output(Node, Value), Nodes), 373: put(nodes, Nodes1). 374: 375: reset_node(I) -> 376: Config = get(config), 377: Nodes = get(nodes), 378: L = repository:get_rank_list(config:get_num_rounds(Config), I), 379: lists:foreach(fun(Path) -> 380: Node = get_node(Path), 381: Nodes1 = dict:store(Path, node:set_output_as_input(Node), Nodes), 382: put(nodes, Nodes1) 383: end, L). 384: 385: 386: increment_count(Child) -> 387: Counts = get(counts), 388: Node = get_node(Child), 389: Count = dict:fetch(node:get_output(Node), Counts) + 1, 390: Counts1 = dict:store(node:get_output(Node), Count, Counts), 391: put(counts, Counts1). 392: 393: find_majority(Path) -> 394: Counts = dict:new(), 395: Counts1 = dict:store(?ONE, 0, Counts), 396: Counts2 = dict:store(?ZERO, 0, Counts1), 397: Counts3 = dict:store(?UNKNOWN, 0, Counts2), 398: put(counts, Counts3), 399: L = repository:get_children_path(Path), 400: N = length(L), 401: lists:foreach(fun(Child) -> 402: increment_count(Child) end, L), 403: Counts4 = get(counts), 404: OneCount = dict:fetch(?ONE, Counts4), 405: ZeroCount = dict:fetch(?ZERO, Counts4), 406: 407: if 408: OneCount > ( N / 2 ) -> 409: ?ONE; 410: ZeroCount > ( N / 2 ) -> 411: ?ZERO; 412: OneCount == ZeroCount andalso OneCount == ( N / 2 ) -> 413: strategy:get_default(); 414: true -> 415: ?UNKNOWN 416: end. 417: 418: 419: loop(Id, Config) -> 420: receive 421: {_From, init, AllPids} -> 422: init(Id, Config, AllPids), 423: loop(Id, Config); 424: {_From, {receive_message, Path, Node}} -> 425: receive_message(Path, Node), 426: %From ! {self(), done}, 427: loop(Id, Config); 428: {_From, {send_messages, Round, Id}} -> 429: psend_messages(Round, Id), 430: %From ! {self(), done}, 431: loop(Id, Config); 432: {From, decide} -> 433: From ! {self(), decide()}, 434: loop(Id, Config); 435: {From, is_faulty} -> 436: From ! {self(), strategy:is_faulty(Config, Id)}, 437: loop(Id, Config); 438: {From, is_source} -> 439: From ! {self(), config:get_source(Config) == Id}, 440: loop(Id, Config); 441: {From, {find_majority, Path}} -> 442: From ! {self(), find_majority(Path)}, 443: loop(Id, Config); 444: Any -> 445: io:format("Unexpected ~p~n", [Any]), 446: loop(Id, Config) 447: end. 448: 449: 453: test_process() -> 454: counter:start(), 455: repository:start(), 456: Config = config:new(3, 5, 6), 457: Pid = start(0, Config), 458: init(Pid, [Pid, 1, 2, 3, 4, 5, 6]), 459: Path = "mypath", 460: Node = node:new(), 461: receive_message(Pid, Path, Node), 462: send_messages(Pid, 0, 0), 463: ?UNKNOWN = decide(Pid), 464: ?ONE = find_majority(Pid, Path), 465: false = is_faulty(Pid), 466: false = is_source(Pid). 467: 468: 469: 470: 471: 472: 476: -module(repository). 477: -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]). 478: -include("byz_general_test.hrl"). 479: 480: 481: 482: 486: start() -> 487: register(repository, spawn(fun() -> init_loop() end)). 488: 489: 490: init_loop() -> 491: Children = ets:new(children, [set]), 492: PathsByRank = ets:new(path_ranks, [set]), 493: loop(Children, PathsByRank). 494: 495: get_children_path(Path) -> 496: lib_util:rpc(repository, {get_children_path, Path}). 497: 498: get_rank_list(Rank, Source) -> 499: lib_util:rpc(repository, {get_rank_list, Rank, Source}). 500: 501: generate_children(Config) -> 502: repository ! {self(), generate_children, Config}. 503: 504: reset() -> 505: repository ! {self(), reset}. 506: 507: die() -> 508: repository ! {self(), die, "Exiting"}. 509: 515: 519: table_lookup(Tab, Key) -> 520: Result = ets:lookup(Tab, Key), 521: case Result of 522: [] -> 523: []; 524: error -> 525: []; 526: {ok, Value} -> 527: lists:reverse(Value); 528: [{Key, Value}] -> 529: lists:reverse(Value) 530: end. 531: 532: table_insert(Tab, Key, Value) -> 533: NewValue = case table_lookup(Tab, Key) of 534: [] -> 535: [Value]; 536: L -> 537: [Value|L] 538: end, 539: ets:insert(Tab, {Key, NewValue}). 540: 541: 542: 546: set_children_path(Children, PathKey, PathValue) -> 547: table_insert(Children, PathKey, PathValue). 548: 549: get_children_path(Children, Path) -> 550: table_lookup(Children, Path). 551: 552: 553: 554: 558: get_rank_list(PathsByRank, Rank, Source) -> 559: table_lookup(PathsByRank, {Rank, Source}). 560: 561: 562: set_rank_list(PathsByRank, Rank, Source, Path) -> 563: Key = {Rank, Source}, 564: table_insert(PathsByRank, Key, Path). 565: 566: 567: generate_children(Config, Children, PathsByRank) -> 568: generate_children( 569: Config, Children, PathsByRank, 570: config:get_source(Config), 571: [], 572: "", 573: 0). 574: 575: generate_children(Config, Children, PathsByRank, Source, Ids, Path, Rank) -> 576: Ids1 = [Source|Ids], 577: Path1 = Path ++ integer_to_list(Source), 578: set_rank_list(PathsByRank, Rank, Source, Path1), 579: %%% 580: Rounds = config:get_num_rounds(Config), 581: if 582: Rank < Rounds -> 583: IdsToProcess = lists:seq(0, config:get_num_procs(Config)-1) -- Ids1, 584: lists:foreach( 585: fun(Source1) -> generate_children(Config, Children, PathsByRank, Source1, Ids1, Path1, Rank+1), 586: set_children_path(Children, Path1, Path1 ++ integer_to_list(Source1)) 587: end, IdsToProcess); 588: true -> 589: true 590: end, 591: Children. 592: 593: 594: loop(Children, PathsByRank) -> 595: receive 596: {From, {get_rank_list, Rank, Source}} -> 597: From ! {repository, get_rank_list(PathsByRank, Rank, Source)}, 598: loop(Children, PathsByRank); 599: {From, {get_children_path, Path}} -> 600: From ! {repository, get_children_path(Children, Path)}, 601: loop(Children, PathsByRank); 602: {_From, reset} -> 603: ets:delete_all_objects(Children), 604: ets:delete_all_objects(PathsByRank), 605: loop(Children, PathsByRank); 606: {_From, generate_children, Config} -> 607: generate_children(Config, Children, PathsByRank), 608: loop(Children, PathsByRank); 609: {_From, die, Reason} -> 610: ets:delete(Children), 611: ets:delete(PathsByRank), 612: exit(Reason); 613: Any -> 614: io:format("Unexpected message~p~n", [Any]) 615: end. 616: 617: 618: 720: 724: benchmark_byz_general() -> 725: engine:init(), 726: benchmark_byz_general(5). 727: 728: benchmark_byz_general(N) when N < 100 -> 729: benchmark_byz_general(N, 10), 730: benchmark_byz_general(N+10); 731: benchmark_byz_general(N) when N >= 100 -> 732: true. 733: 734: benchmark_byz_general(N, M) when M < 100 -> 735: statistics(runtime), 736: statistics(wall_clock), 737: Source = round(N / 3), 738: io:format("Starting|~p|~p|~p~n", [M, N, Source]), 739: engine:start(Source, N, M), 740: engine:run(), 741: {_, RT} = statistics(runtime), 742: {_, WC} = statistics(wall_clock), 743: io:format("Finished|~p|~p|~p|~p|~p|~p~n", [M, N, Source, counter:get_value(), RT, WC]), 744: engine:reset(), 745: benchmark_byz_general(N, M+10); 746: benchmark_byz_general(_N, M) when M >= 100 -> 747: true. 748: 749:
I merged the results with following script:
1: class Stats 2: attr_accessor :processes 3: attr_accessor :rounds 4: attr_accessor :messages 5: attr_accessor :java_time 6: attr_accessor :erlang_time 7: attr_accessor :cpp_time 8: attr_accessor :ruby_time 9: def initialize(procs, rounds, msgs) 10: @processes = procs.to_i 11: @rounds = rounds.to_i 12: @messages = msgs.to_i 13: end 14: def key 15: "#{@processes}/#{@rounds}" 16: end 17: def to_s 18: "#{@processes},#{rounds},#{@messages},#{@cpp_time},#{@ruby_time},#{@java_time},#{@erlang_time}" 19: end 20: end 21: 22: stats = {} 23: 24: files = ["cpp.out", "ruby.out", "java.out", "erlang.out"] 25: count_setter = ["cpp_time=", "ruby_time=", "java_time=", "erlang_time="] 26: for i in 0...files.length 27: File.open(files[i], "r").readlines.each do |line| 28: if line =~ /Finish/ 29: t = line.split("|") 30: stat = Stats.new(t[2], t[1], t[4]) 31: time = t[5].to_i 32: stat.send(count_setter[i], time) 33: stats[stat.key] = stat 34: end 35: end 36: end 37: 38: 39: puts "Processes,Rounds,Messages,CPP Time,Ruby Time,Java Time,Erlang Time" 40: stats.values.sort_by {|stat| stat.processes * stat.rounds}.each do |stat| 41: puts stat 42: end 43:
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 indpendent processes and communication because that would have required a lot more effort.