Add read repair

coderoshi · coderoshi · commit ac454a19d0e1 · 2013-02-20T22:03:14.000-08:00
diff --git a/09-repair/client.rb b/09-repair/client.rb
@@ -4,10 +4,19 @@
 ctx = ZMQ::Context.new
 req = ctx.socket(ZMQ::REQ)
 req.connect("tcp://127.0.0.1:2200")
+req.send("put 0 foo {\"B\":1} baz") && req.recv
+req.close
+
+req = ctx.socket(ZMQ::REQ)
+req.connect("tcp://127.0.0.1:2201")
+req.send("put 0 foo {} qux") && req.recv
+
+# trigger read repair
+puts req.send("get 2 foo") && req.recv
+
+sleep 1
 
-req.send("put_counter 1 foo +1") && req.recv
-req.send("put_counter 1 foo +2") && req.recv
-req.send("put_counter 2 foo +1") && req.recv
-puts req.send("get_counter 2 foo") && req.recv
+# read repair should be complete
+puts req.send("get 2 foo") && req.recv
 
 req.close
diff --git a/09-repair/node.rb b/09-repair/node.rb
@@ -132,6 +132,7 @@ def do_get(key, n=1, crdt=nil)
       results.flatten!
       results.uniq! {|o| o && o.value }
       results.compact!
+      repair(key, n)
       begin
         # we get the most recent value if we can resolve this
         [results.sort.first]
@@ -146,15 +147,34 @@ def do_get(key, n=1, crdt=nil)
   end
 
   def replicate(message, key, n)
-    list = @ring.pref_list(key, n)
-    list -= [@name]
+    list = @ring.pref_list(key, n) - [@name]
     results = []
     while replicate_node = list.shift
       results << remote_call(replicate_node, message)
     end
     results
   end
 
+  def repair(key, n)
+    list = @ring.pref_list(key, n) - [@name]
+    puts "Repairing #{key}"
+    list.map do |replicate_node|
+      Thread.new do
+        result = remote_call(replicate_node, "get 0 #{key}")
+        remote_objs = NodeObject.deserialize(result)
+        if remote_objs != 'null'
+          # if local is nil or descends, update local
+          local = @data[@ring.hash(key)]
+          vclock = local && local.first.vclock
+          descends = remote_objs.find{|o| o.vclock.descends_from?(vclock)}
+          if vclock == nil || descends
+            @data[@ring.hash(key)] = nos
+          end
+        end
+      end
+    end
+  end
+
   def remote_call(remote_name, message)
     puts "#{remote_name} <= #{message}"
     remote_port = config(remote_name)["port"]
