Fix vclock/sibling code, extract mapreduce

coderoshi · coderoshi · commit 7eb466b96f22 · 2013-02-11T21:42:22.000-08:00
diff --git a/client.rb b/client.rb
@@ -7,7 +7,7 @@
 # How do we live beyond node failure? N Replication
 # How do we know which value is correct? Add Vector Clocks
 
-
+# CRDT
 # Use PUSH/PULL for AAE
 # Map/Reduce should be the last thing
 # TODO: Make add-ons modules
@@ -16,7 +16,8 @@
 # Replace pub/sub ring-leader with req/res gossip protocol
 # Allow sloppy quorums and data transfer of downed nodes with hinted handoff
 
-port = 4000
+# port = 4000
+port = 2201
 ctx = ZMQ::Context.new
 req = ctx.socket(ZMQ::REQ)
 req.connect("tcp://127.0.0.1:#{port}")
@@ -37,8 +38,10 @@
 # resp = req.send("put 2 key {\"A:2201\":2,\"B:2201\":2} hello") && req.recv
 
 # Counters
-req.send("put 1 key {\"A:2201\":1} +1") && req.recv
-req.send("put 1 key {\"B:2202\":1} +1") && req.recv
+# req.send("put 1 key {\"A:2201\":1} +1") && req.recv
+# req.send("put 1 key {\"B:2202\":1} +1") && req.recv
+req.send("put 1 key {} +1") && req.recv
+
 vals = req.send("get 2 key") && req.recv
 p JSON.parse(vals)
 p JSON.parse(vals).reduce(0){|sum,v| sum + v['value'].to_i}
diff --git a/mapreduce.rb b/mapreduce.rb
@@ -0,0 +1,61 @@
+
+class Map
+  def initialize(func_str, data)
+    @data = data
+    @func = func_str
+  end
+
+  def call
+    eval(@func, binding)
+  end
+
+  # calls given map block for every value
+  def map
+    @results = []
+    @data.each {|k,v| @results += yield k,v }
+    @results
+  end
+end
+
+class Reduce
+  def initialize(func_str, results)
+    @results = results
+    @func = func_str
+  end
+
+  def call
+    eval(@func, binding)
+  end
+
+  # calls given reduce block for every value
+  def reduce
+    yield @results
+  end
+end
+
+
+
+module Mapreduce
+  def map(socket, payload)
+    socket.send( Map.new(payload, @data).call.to_s )
+  end
+
+  def mr(socket, payload)
+    map_func, reduce_func = payload.split(/\;\s+reduce/, 2)
+    reduce_func = "reduce#{reduce_func}"
+    socket.send( Reduce.new(reduce_func, remote_maps(map_func)).call.to_s )
+  end
+
+  # run in parallel, then join results
+  def remote_maps(map_func)
+    workers, results = [], []
+    @nodes.each do |node|
+      workers << Thread.new do
+        res = remote_call(node, "map #{map_func}")
+        results += eval(res)
+      end
+    end
+    workers.each{|w| w.join}
+    results
+  end
+end
diff --git a/merkle.rb b/merkle.rb
@@ -1,10 +1,7 @@
 require 'digest/sha1'
 
 class MerkleTree
-  attr :nodes
-  attr :blocks
-  attr :block_size
-  attr :root
+  attr :nodes, :blocks, :block_size, :root
 
   def initialize(block_size=1024)
     @block_size = block_size
@@ -113,7 +110,12 @@ def add_level(this_level, next_level)
   end
 end
 
-mt = MerkleTree.new()
-ARGF.each { |line| mt.add_line(line) }
-mt.finish
-mt.output
+### run tests
+if __FILE__ == $0
+
+  mt = MerkleTree.new()
+  ARGF.each { |line| mt.add_line(line) }
+  mt.finish
+  mt.output
+
+end
diff --git a/node.rb b/node.rb
